Skip to content

Python and the GitHub API: Building a Small Client

Lesson 10 of 10Intermediate → Advanced10 min readGitHub Engineering · GitHub APIVerified: Python 3.12 with requests 2.x; GitHub REST API version 2022-11-28, August 2026

At some point shell scripting stops being the right tool — usually when you are building JSON in string concatenation, or when you want tests.

This lesson builds a small GitHub client in Python: enough structure to be maintainable, not so much that it becomes a framework.

Use requests, or httpx if you want async. Both are current, widely used and well maintained.

Wrapper libraries exist that model GitHub’s API as Python objects. They can be convenient and they carry a real risk: a wrapper lags the API, and when GitHub adds an endpoint you need, you are waiting for someone else. Several popular GitHub wrappers have been abandoned over the years, and inheriting one is worse than having written eighty lines yourself.

The API is a well-documented HTTP interface. Calling it directly means you are never blocked, and the code below is not much longer than learning a wrapper’s conventions would be.

Terminal window
python3 -m venv .venv
source .venv/bin/activate
pip install requests
requests>=2.31

The token comes from the environment, never from the source:

Terminal window
export GITHUB_TOKEN="YOUR_TOKEN_HERE"
github_client/client.py
"""A small, honest GitHub REST client."""
from __future__ import annotations
import logging
import os
import time
from typing import Any, Iterator
import requests
log = logging.getLogger(__name__)
API_ROOT = "https://api.github.com"
API_VERSION = "2022-11-28"
class GitHubError(RuntimeError):
"""An API call failed in a way the caller should handle."""
def __init__(self, message: str, status: int, body: Any = None) -> None:
super().__init__(message)
self.status = status
self.body = body
class GitHubClient:
def __init__(
self,
token: str | None = None,
*,
user_agent: str = "mga-example/1.0",
timeout: float = 10.0,
max_retries: int = 3,
) -> None:
token = token or os.environ.get("GITHUB_TOKEN")
if not token:
raise GitHubError("GITHUB_TOKEN is not set", status=0)
self.timeout = timeout
self.max_retries = max_retries
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {token}",
"Accept": "application/vnd.github+json",
"X-GitHub-Api-Version": API_VERSION,
"User-Agent": user_agent,
})
def request(self, method: str, path: str, **kwargs: Any) -> requests.Response:
url = path if path.startswith("http") else f"{API_ROOT}/{path.lstrip('/')}"
for attempt in range(1, self.max_retries + 1):
response = self.session.request(
method, url, timeout=self.timeout, **kwargs
)
if self._should_wait(response) and attempt < self.max_retries:
delay = self._retry_delay(response, attempt)
log.warning("rate limited; sleeping %.1fs (attempt %d)", delay, attempt)
time.sleep(delay)
continue
if response.status_code >= 500 and attempt < self.max_retries:
delay = 2 ** attempt
log.warning("server error %s; retrying in %ds", response.status_code, delay)
time.sleep(delay)
continue
if not response.ok:
raise GitHubError(
f"{method} {url} failed: {response.status_code}",
status=response.status_code,
body=self._safe_json(response),
)
return response
raise GitHubError(f"{method} {url} failed after {self.max_retries} attempts", status=0)
def get(self, path: str, **kwargs: Any) -> Any:
return self.request("GET", path, **kwargs).json()
def post(self, path: str, json: dict[str, Any]) -> Any:
return self.request("POST", path, json=json).json()
def patch(self, path: str, json: dict[str, Any]) -> Any:
return self.request("PATCH", path, json=json).json()
def delete(self, path: str) -> None:
self.request("DELETE", path)
def paginate(self, path: str, **kwargs: Any) -> Iterator[dict[str, Any]]:
"""Yield every item across all pages, following the Link header."""
params = dict(kwargs.pop("params", {}))
params.setdefault("per_page", 100)
url: str | None = path
while url:
response = self.request("GET", url, params=params, **kwargs)
payload = response.json()
if isinstance(payload, list):
yield from payload
else:
yield payload
url = response.links.get("next", {}).get("url")
params = {} # the next URL already carries them
def rate_limit(self) -> dict[str, Any]:
return self.get("rate_limit")["resources"]["core"]
@staticmethod
def _should_wait(response: requests.Response) -> bool:
if response.status_code == 429:
return True
return (
response.status_code == 403
and response.headers.get("x-ratelimit-remaining") == "0"
)
@staticmethod
def _retry_delay(response: requests.Response, attempt: int) -> float:
retry_after = response.headers.get("retry-after")
if retry_after and retry_after.isdigit():
return float(retry_after)
reset = response.headers.get("x-ratelimit-reset")
if reset and reset.isdigit():
return max(0.0, float(reset) - time.time()) + 1.0
return float(2 ** attempt)
@staticmethod
def _safe_json(response: requests.Response) -> Any:
try:
return response.json()
except ValueError:
return response.text[:500]

Several decisions in there are worth stating explicitly.

timeout is mandatory and defaulted. requests has no default timeout — omit it and a hung connection blocks forever. In a scheduled job that means a stuck run rather than a failed one, which is much harder to notice.

Rate limits and server errors are retried; client errors are not. A 422 will fail identically however many times you try it. Retrying it wastes time and hides the real problem.

paginate follows the Link header rather than incrementing a page number, and clears the params after the first request because the next URL already carries them. This is the correctness point from the REST lesson, implemented once.

Errors carry the parsed body. A 422’s errors array names the offending field. Discarding it turns a precise message into “something went wrong”.

Splitting by resource keeps each file small and testable.

github_client/repositories.py
from __future__ import annotations
from typing import Any, Iterator
from .client import GitHubClient
def get(client: GitHubClient, owner: str, repo: str) -> dict[str, Any]:
return client.get(f"repos/{owner}/{repo}")
def list_for_org(client: GitHubClient, org: str, *, include_forks: bool = False) -> Iterator[dict]:
params = {"type": "all" if include_forks else "sources"}
yield from client.paginate(f"orgs/{org}/repos", params=params)
def update_settings(client: GitHubClient, owner: str, repo: str, **settings: Any) -> dict[str, Any]:
return client.patch(f"repos/{owner}/{repo}", json=settings)
github_client/issues.py
from __future__ import annotations
from typing import Any, Iterator
from .client import GitHubClient
def list_open(client: GitHubClient, owner: str, repo: str) -> Iterator[dict[str, Any]]:
"""Open issues only — the endpoint also returns pull requests."""
for item in client.paginate(f"repos/{owner}/{repo}/issues", params={"state": "open"}):
if "pull_request" not in item:
yield item
def create(
client: GitHubClient, owner: str, repo: str, title: str, body: str = "",
labels: list[str] | None = None,
) -> dict[str, Any]:
payload: dict[str, Any] = {"title": title, "body": body}
if labels:
payload["labels"] = labels
return client.post(f"repos/{owner}/{repo}/issues", json=payload)
def add_labels(client: GitHubClient, owner: str, repo: str, number: int, labels: list[str]) -> Any:
return client.post(f"repos/{owner}/{repo}/issues/{number}/labels", json={"labels": labels})
def close(client: GitHubClient, owner: str, repo: str, number: int, reason: str = "completed") -> Any:
return client.patch(
f"repos/{owner}/{repo}/issues/{number}",
json={"state": "closed", "state_reason": reason},
)

The pull request filter lives in list_open rather than in every caller. That is the main argument for a client at all: encode the API’s sharp edges once.

github_client/pull_requests.py
from __future__ import annotations
from typing import Any, Iterator
from .client import GitHubClient
def list_open(client: GitHubClient, owner: str, repo: str) -> Iterator[dict[str, Any]]:
yield from client.paginate(f"repos/{owner}/{repo}/pulls", params={"state": "open"})
def get(client: GitHubClient, owner: str, repo: str, number: int) -> dict[str, Any]:
return client.get(f"repos/{owner}/{repo}/pulls/{number}")
def merge(
client: GitHubClient, owner: str, repo: str, number: int,
*, expected_sha: str, method: str = "squash",
) -> dict[str, Any]:
"""Merge only if the head is still expected_sha; raises on a race."""
return client.request(
"PUT",
f"repos/{owner}/{repo}/pulls/{number}/merge",
json={"merge_method": method, "sha": expected_sha},
).json()

expected_sha is keyword-only and has no default, so the guarded merge from Pull Request Automation cannot be skipped by accident.

github_client/__init__.py
from .client import GitHubClient, GitHubError
__all__ = ["GitHubClient", "GitHubError"]
report.py
import logging
import sys
from github_client import GitHubClient, GitHubError
from github_client import issues, repositories
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s %(levelname)s %(message)s",
)
log = logging.getLogger("report")
def main(owner: str, repo: str) -> int:
client = GitHubClient(user_agent="mga-report/1.0")
budget = client.rate_limit()
log.info("rate limit: %s/%s remaining", budget["remaining"], budget["limit"])
if budget["remaining"] < 100:
log.error("insufficient rate limit budget")
return 75 # EX_TEMPFAIL — a retry later may succeed
details = repositories.get(client, owner, repo)
log.info("%s%s stars", details["full_name"], details["stargazers_count"])
counts: dict[str, int] = {}
total = 0
for issue in issues.list_open(client, owner, repo):
total += 1
for label in issue["labels"]:
counts[label["name"]] = counts.get(label["name"], 0) + 1
log.info("%d open issues (excluding pull requests)", total)
for name, count in sorted(counts.items(), key=lambda kv: -kv[1])[:10]:
print(f"{count:5d} {name}")
return 0
if __name__ == "__main__":
if len(sys.argv) != 3:
print("usage: report.py OWNER REPO", file=sys.stderr)
raise SystemExit(2)
try:
raise SystemExit(main(sys.argv[1], sys.argv[2]))
except GitHubError as error:
log.error("GitHub API error (%s): %s", error.status, error)
raise SystemExit(1)

Note the separation: logs go to stderr, data goes to stdout. That keeps the script pipeable, which is the same principle as the shell scripting lesson.

import logging
import re
TOKEN_PATTERN = re.compile(r"(gh[pousr]_[A-Za-z0-9]{16,}|Bearer\s+\S+)")
class RedactingFilter(logging.Filter):
def filter(self, record: logging.LogRecord) -> bool:
if isinstance(record.msg, str):
record.msg = TOKEN_PATTERN.sub("[REDACTED]", record.msg)
return True
logging.getLogger().addFilter(RedactingFilter())

What it doesRedacts anything token-shaped from log output.

Why we run itDebug logging that dumps request headers will print your Authorization header. A filter at the logging layer is a backstop that does not depend on remembering at every call site.

Expected resultLog lines with credentials replaced by a placeholder.

Never log request headers or full response bodies at INFO. If you need them while debugging, gate them behind a flag and redact.

The client is testable precisely because it is thin. Mock at the HTTP layer, not the client layer — that way your tests exercise pagination, retry and error handling rather than skipping them.

import responses # pip install responses
from github_client import GitHubClient
from github_client import issues
@responses.activate
def test_list_open_excludes_pull_requests(monkeypatch):
monkeypatch.setenv("GITHUB_TOKEN", "test-token-not-real")
responses.add(
responses.GET,
"https://api.github.com/repos/o/r/issues",
json=[
{"number": 1, "title": "A bug", "labels": []},
{"number": 2, "title": "A PR", "labels": [], "pull_request": {"url": "..."}},
],
status=200,
)
client = GitHubClient()
found = list(issues.list_open(client, "o", "r"))
assert [i["number"] for i in found] == [1]

That single test pins the behaviour most likely to be got wrong.

REST covers most needs; some queries are far cheaper in GraphQL. Adding it is a single method rather than a second client:

github_client/graphql.py
from __future__ import annotations
from typing import Any
from .client import GitHubClient, GitHubError
GRAPHQL_URL = "https://api.github.com/graphql"
def query(client: GitHubClient, document: str, **variables: Any) -> dict[str, Any]:
"""Run a GraphQL query. Raises on errors, which arrive with HTTP 200."""
response = client.request(
"POST", GRAPHQL_URL, json={"query": document, "variables": variables}
)
payload = response.json()
if "errors" in payload:
messages = "; ".join(e.get("message", "?") for e in payload["errors"])
raise GitHubError(f"GraphQL error: {messages}", status=200, body=payload)
return payload["data"]

The errors check is the whole reason this needs a wrapper. GraphQL returns HTTP 200 for most failures, so raise_for_status() passes and the caller then fails confusingly on a null field. Checking once, centrally, means every call site gets it right.

Cursor pagination as a generator:

def paginate(client: GitHubClient, document: str, path: list[str], **variables: Any):
"""Yield nodes across pages. `path` locates the connection in the response."""
cursor = None
while True:
data = query(client, document, endCursor=cursor, **variables)
node = data
for key in path:
node = node[key]
yield from node["nodes"]
info = node["pageInfo"]
if not info["hasNextPage"]:
return
cursor = info["endCursor"]

Used as:

REPOS = """
query($endCursor: String) {
viewer {
repositories(first: 100, after: $endCursor) {
pageInfo { hasNextPage endCursor }
nodes { nameWithOwner isPrivate }
}
}
}
"""
for repo in paginate(client, REPOS, ["viewer", "repositories"]):
print(repo["nameWithOwner"])

Anything that writes should be able to describe what it would do. Threading a flag through every function is tedious; putting it on the client is not:

class GitHubClient:
def __init__(self, token: str | None = None, *, dry_run: bool = False) -> None:
# ... existing setup ...
self.dry_run = dry_run
def request(self, method: str, path: str, **kwargs: Any) -> requests.Response:
if self.dry_run and method.upper() not in {"GET", "HEAD"}:
log.info("DRY RUN %s %s %s", method, path, kwargs.get("json", ""))
simulated = requests.Response()
simulated.status_code = 200
simulated._content = b"{}"
return simulated
# ... existing request, retry and error handling ...
raise NotImplementedError

Intercepting at the client means every write is covered, including ones added later by someone who never read this lesson. Reads still happen, so the dry run exercises the real query logic and only the mutations are simulated — which is exactly the behaviour you want.

client = GitHubClient(dry_run=os.environ.get("DRY_RUN", "true") == "true")

Defaulting to true means the destructive mode requires an explicit opt-in.

The client retries when limited. A long-running job should also check before starting, and report as it goes:

def check_budget(client: GitHubClient, needed: int) -> None:
core = client.rate_limit()
if core["remaining"] < needed:
reset = datetime.fromtimestamp(core["reset"], tz=timezone.utc)
raise GitHubError(
f"need {needed} requests, {core['remaining']} remaining until {reset:%H:%M UTC}",
status=0,
)
log.info("rate limit: %s/%s remaining", core["remaining"], core["limit"])

Estimating needed is usually straightforward: a sweep over N repositories making two calls each needs roughly 2N plus pagination. Being approximately right is enough — the point is to fail at the start rather than three hundred repositories in.

Mock at the HTTP layer so pagination, retry and error handling are exercised rather than skipped.

import responses
from github_client import GitHubClient, GitHubError
from github_client import issues
@responses.activate
def test_pagination_follows_link_header(monkeypatch):
monkeypatch.setenv("GITHUB_TOKEN", "test-token-not-real")
responses.add(
responses.GET, "https://api.github.com/repos/o/r/issues",
json=[{"number": 1, "labels": []}], status=200,
headers={"Link": '<https://api.github.com/repos/o/r/issues?page=2>; rel="next"'},
)
responses.add(
responses.GET, "https://api.github.com/repos/o/r/issues",
json=[{"number": 2, "labels": []}], status=200,
)
found = list(issues.list_open(GitHubClient(), "o", "r"))
assert [i["number"] for i in found] == [1, 2]
@responses.activate
def test_client_errors_are_not_retried(monkeypatch):
monkeypatch.setenv("GITHUB_TOKEN", "test-token-not-real")
responses.add(
responses.POST, "https://api.github.com/repos/o/r/issues",
json={"message": "Validation Failed",
"errors": [{"field": "title", "code": "missing_field"}]},
status=422,
)
client = GitHubClient(max_retries=3)
try:
issues.create(client, "o", "r", title="")
except GitHubError as error:
assert error.status == 422
assert error.body["errors"][0]["field"] == "title"
assert len(responses.calls) == 1 # not retried

The second test pins behaviour that is easy to get wrong and expensive in production: a client that retries 422 three times turns one clear error into three, with a delay, and still fails.

Once several scripts share the client, make it installable rather than copied:

pyproject.toml
[project]
name = "github-client"
version = "0.1.0"
requires-python = ">=3.11"
dependencies = ["requests>=2.31"]
[project.optional-dependencies]
dev = ["pytest>=8", "responses>=0.25", "mypy>=1.8"]
[build-system]
requires = ["setuptools>=68"]
build-backend = "setuptools.build_meta"
Terminal window
pip install -e '.[dev]'
pytest
mypy github_client

An editable install means scripts import it rather than sitting beside it, and there is one copy to fix when GitHub changes something. Copies of a client diverging across four repositories is the failure this prevents.

Deliberate omissions, because the temptation to add them is strong:

No ORM-style objects. Dictionaries match the API and need no maintenance when GitHub adds fields.

No caching layer. Add one when a measurement says you need it.

No async. Unless you have measured that request latency dominates, requests is simpler.

No auto-generated models. They are a maintenance burden for a client this size.

The goal is code a colleague can read in ten minutes and change without archaeology.

Everything assembled into something that would survive being scheduled:

#!/usr/bin/env python3
"""Report open issues with no assignee, grouped by label."""
from __future__ import annotations
import argparse
import logging
import sys
from collections import Counter
from github_client import GitHubClient, GitHubError
from github_client import issues
log = logging.getLogger("triage")
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("repo", help="OWNER/REPO")
parser.add_argument("--min-budget", type=int, default=200,
help="refuse to run below this many remaining requests")
parser.add_argument("--verbose", action="store_true")
return parser.parse_args()
def main() -> int:
args = parse_args()
logging.basicConfig(
level=logging.DEBUG if args.verbose else logging.INFO,
format="%(asctime)s %(levelname)s %(message)s",
stream=sys.stderr,
)
if "/" not in args.repo:
log.error("expected OWNER/REPO, got %r", args.repo)
return 2
owner, repo = args.repo.split("/", 1)
client = GitHubClient(user_agent="mga-triage/1.0")
budget = client.rate_limit()
log.info("rate limit: %s/%s remaining", budget["remaining"], budget["limit"])
if budget["remaining"] < args.min_budget:
log.error("insufficient budget: %s remaining", budget["remaining"])
return 75 # EX_TEMPFAIL
labels: Counter[str] = Counter()
unassigned = 0
for issue in issues.list_open(client, owner, repo):
if issue["assignees"]:
continue
unassigned += 1
for label in issue["labels"]:
labels[label["name"]] += 1
log.info("%d unassigned open issue(s)", unassigned)
for name, count in labels.most_common(10):
print(f"{count:5d} {name}") # data on stdout
return 0
if __name__ == "__main__":
try:
raise SystemExit(main())
except GitHubError as error:
log.error("GitHub API error (%s): %s", error.status, error)
raise SystemExit(1)
except KeyboardInterrupt:
raise SystemExit(130)

Points worth copying: argument validation before any network call, a budget check that fails with EX_TEMPFAIL so a scheduler knows to retry later, logs to stderr and data to stdout, and distinct exit codes for usage error, API failure and interruption.

The client is small enough that type hints are worth having and cheap to add.

Terminal window
mypy --strict github_client

--strict will complain about the dict[str, Any] returns, which is the honest state of things — GitHub’s responses are dictionaries and modelling every field is a maintenance burden the earlier section deliberately avoided. Typing the boundaries while leaving payloads as Any is a reasonable middle position:

def get_repo(client: GitHubClient, owner: str, repo: str) -> dict[str, Any]:
...
def list_open(client: GitHubClient, owner: str, repo: str) -> Iterator[dict[str, Any]]:
...

Callers get the shape of the function; nobody maintains a model of GitHub’s schema. If a particular response shape matters enough, a TypedDict for the handful of fields you use is a better investment than a full model.

For a client making dozens of requests where latency dominates, httpx with asyncio is a real improvement:

import asyncio
import httpx
async def fetch_all(repos: list[str], token: str) -> list[dict]:
headers = {
"Authorization": f"Bearer {token}",
"Accept": "application/vnd.github+json",
"X-GitHub-Api-Version": "2022-11-28",
}
limits = httpx.Limits(max_connections=8)
async with httpx.AsyncClient(headers=headers, timeout=10.0, limits=limits) as http:
tasks = [http.get(f"https://api.github.com/repos/{r}") for r in repos]
responses = await asyncio.gather(*tasks, return_exceptions=True)
return [r.json() for r in responses if isinstance(r, httpx.Response) and r.is_success]

max_connections=8 is the important line. Unbounded concurrency against GitHub will hit the hundred-concurrent secondary limit and produce failures that look like server errors.

The caveat from earlier still applies: concurrency spends the same rate-limit budget faster. It helps when latency is the constraint and hurts when the budget is.

Reach for this when you have measured that requests are the bottleneck. Adding async to a script making twenty calls complicates it for no gain.

No timeout. requests has no default; a hung call blocks forever.

Retrying 4xx. Wastes time and hides real errors.

Incrementing page numbers instead of following Link. Fragile and occasionally wrong.

Discarding the error body. Loses the errors array that names the problem.

Logging headers. Prints your token.

Forgetting the pull request filter. Inflated Issue counts.

A wrapper library that lags the API. Blocks you when you need something new.

Logging to stdout. Corrupts data for anything downstream.

  1. Build the package as laid out, in a virtual environment.
  2. Run report.py against a public repository and confirm the label counts.
  3. Compare the Issue total with open_issues_count and explain the difference.
  4. Set an invalid token and confirm the failure is a clear GitHubError with status 401.
  5. Set the timeout to 0.001 and confirm the exception surfaces rather than hanging.
  6. Write the pagination test above and confirm it passes.
  7. Add a --dry-run flag to any function that writes.

Step 3 is the one that turns a lesson into a habit — the discrepancy is real on any active repository.

  • Call the API directly with requests; wrapper libraries lag and get abandoned.
  • Timeouts are mandatory, because requests has no default.
  • Retry rate limits and 5xx; never retry 4xx.
  • Follow the Link header for pagination and implement it once, in the client.
  • Encode the API’s sharp edges — such as the pull request overlap — in the client, not in callers.
  • Keep credentials out of code and out of logs, with redaction as a backstop.
  • Logs to stderr, data to stdout.
  • Leave out the ORM, the cache and the async until something measured demands them.

Call the API directly with requests. Wrapper libraries lag GitHub’s changes and several popular ones have been abandoned; the client here is under two hundred lines and you can fix it yourself.

The details separating working from robust: a timeout on every request, retries for 5xx and rate limits but never 4xx, pagination that follows Link, the error body preserved so a 422 names the offending field, and the API’s sharp edges — the pull request overlap in the Issues endpoint above all — encoded once in the client rather than in every caller.

Leave out the ORM, the cache and the async until something you have measured demands them.

Once the client is a package, exposing its scripts as commands is a few lines and changes how people use it.

pyproject.toml
[project.scripts]
gh-triage = "github_client.cli:main"
github_client/cli.py
"""Entry point dispatching to subcommands."""
from __future__ import annotations
import argparse
import logging
import sys
from . import GitHubClient, GitHubError
from . import issues, repositories
log = logging.getLogger("gh-triage")
def cmd_stale(client: GitHubClient, args: argparse.Namespace) -> int:
owner, repo = args.repo.split("/", 1)
count = 0
for issue in issues.list_open(client, owner, repo):
count += 1
print(f"{count} open issues in {args.repo}")
return 0
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(prog="gh-triage")
sub = parser.add_subparsers(dest="command", required=True)
stale = sub.add_parser("stale", help="report stale issues")
stale.add_argument("repo", help="OWNER/REPO")
stale.set_defaults(func=cmd_stale)
args = parser.parse_args(argv)
logging.basicConfig(level=logging.INFO, stream=sys.stderr,
format="%(levelname)s %(message)s")
try:
return args.func(GitHubClient(), args)
except GitHubError as error:
log.error("GitHub API error (%s): %s", error.status, error)
return 1
Terminal window
pip install -e .
gh-triage stale acme/api

The benefit is not convenience — it is that the tool is now installable, versionable and shareable. A colleague runs pip install rather than being told which directory to cd into, and the subcommand structure gives you somewhere obvious to add the next piece of automation.

Professional ToolkitThe fine-grained token permissions matrix, App-vs-PAT guide and gh api recipes are in the Professional Toolkit.