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.
Choosing a library
Section titled “Choosing a library”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.
Setting up
Section titled “Setting up”python3 -m venv .venvsource .venv/bin/activatepip install requestsrequests>=2.31The token comes from the environment, never from the source:
export GITHUB_TOKEN="YOUR_TOKEN_HERE"The core client
Section titled “The core client”"""A small, honest GitHub REST client."""
from __future__ import annotations
import loggingimport osimport timefrom 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”.
Resource modules
Section titled “Resource modules”Splitting by resource keeps each file small and testable.
from __future__ import annotationsfrom typing import Any, Iteratorfrom .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)from __future__ import annotationsfrom typing import Any, Iteratorfrom .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.
from __future__ import annotationsfrom typing import Any, Iteratorfrom .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.
from .client import GitHubClient, GitHubError
__all__ = ["GitHubClient", "GitHubError"]Using it
Section titled “Using it”import loggingimport sys
from github_client import GitHubClient, GitHubErrorfrom 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.
Logging without leaking
Section titled “Logging without leaking”import loggingimport 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.
Testing
Section titled “Testing”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 responsesfrom github_client import GitHubClientfrom github_client import issues
@responses.activatedef 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.
Adding GraphQL
Section titled “Adding GraphQL”REST covers most needs; some queries are far cheaper in GraphQL. Adding it is a single method rather than a second client:
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"])Dry-run as a first-class concept
Section titled “Dry-run as a first-class concept”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 NotImplementedErrorIntercepting 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.
Rate limit awareness
Section titled “Rate limit awareness”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.
Testing the parts that matter
Section titled “Testing the parts that matter”Mock at the HTTP layer so pagination, retry and error handling are exercised rather than skipped.
import responsesfrom github_client import GitHubClient, GitHubErrorfrom github_client import issues
@responses.activatedef 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.activatedef 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 retriedThe 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.
Packaging it
Section titled “Packaging it”Once several scripts share the client, make it installable rather than copied:
[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"pip install -e '.[dev]'pytestmypy github_clientAn 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.
What to leave out
Section titled “What to leave out”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.
A worked script
Section titled “A worked script”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 argparseimport loggingimport sysfrom collections import Counter
from github_client import GitHubClient, GitHubErrorfrom 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.
Type checking
Section titled “Type checking”The client is small enough that type hints are worth having and cheap to add.
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.
Async, and when it earns its place
Section titled “Async, and when it earns its place”For a client making dozens of requests where latency dominates, httpx with asyncio is a real
improvement:
import asyncioimport 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.
Common mistakes
Section titled “Common mistakes”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.
Exercise
Section titled “Exercise”- Build the package as laid out, in a virtual environment.
- Run
report.pyagainst a public repository and confirm the label counts. - Compare the Issue total with
open_issues_countand explain the difference. - Set an invalid token and confirm the failure is a clear
GitHubErrorwith status 401. - Set the timeout to
0.001and confirm the exception surfaces rather than hanging. - Write the pagination test above and confirm it passes.
- Add a
--dry-runflag 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.
What you learned
Section titled “What you learned”- Call the API directly with
requests; wrapper libraries lag and get abandoned. - Timeouts are mandatory, because
requestshas no default. - Retry rate limits and 5xx; never retry 4xx.
- Follow the
Linkheader 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.
The short version
Section titled “The short version”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.
A command-line entry point
Section titled “A command-line entry point”Once the client is a package, exposing its scripts as commands is a few lines and changes how people use it.
[project.scripts]gh-triage = "github_client.cli:main""""Entry point dispatching to subcommands."""
from __future__ import annotations
import argparseimport loggingimport sys
from . import GitHubClient, GitHubErrorfrom . 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 1pip install -e .gh-triage stale acme/apiThe 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.