headgate¶

One base class instead of a stack¶

A production API client usually grows a stack: an HTTP library, a retry library, a rate limiter, a cache, a circuit breaker, a mocking library, and the glue code that makes them agree on deadlines and credentials. headgate is that whole stack, built to work as one.

The usual stack httpx or aiohttp tenacity or stamina aiolimiter hishel or requests-cache pybreaker httpx-sse respx your glue code

→

headgate one dependency, every layer built to work with the others

Each call your client makes passes through every layer, in an order chosen so they cooperate: a cache hit is never retried, a retry spends its own rate turn, and one deadline bounds the lot.

1. Deadlineone budget of time for the whole call
2. Circuit breakerfails fast while the host is down
3. Authcredentials only for your origin
4. Retriessafe re-sends, Retry-After obeyed
5. Cachea hit stops here
6. Rate budgetpaced, and paused when the server asks
7. NetworkHTTP/2, size-capped body
8. Your readera typed result, or a typed error

How headgate compares with 20+ libraries The standards it follows

Install¶

pip install 'headgate[msgspec]'

uv add 'headgate[msgspec]'

The msgspec extra brings the codec the examples use; headgate[pydantic] brings pydantic's, and any other library plugs in as a codec of its own. headgate needs Python 3.14 or newer: endpoint paths are template strings, so each value is escaped into its own path segment by default. It runs on httpx2, pydantic's fork of httpx; see why httpx2.

A real client¶

A client for a few GitHub REST endpoints. The full version, with its tests, is examples/github_client.py.

from http import HTTPStatus

import msgspec

from headgate import AsyncBaseClient, BearerAuth, Expect, HttpOptions, LinkHeader, Rate, RateBudget, nothing
from headgate.msgspec import decoder


class Repository(msgspec.Struct, frozen=True):
    full_name: str
    stargazers_count: int


class Issue(msgspec.Struct, frozen=True):
    number: int
    title: str


REPOSITORY_OR_NONE = Expect(decoder(Repository)).on(HTTPStatus.NOT_FOUND, nothing)
ISSUE_PAGES = LinkHeader(items=lambda page: page)  # each page is a JSON list of issues


class GitHub(AsyncBaseClient):
    def __init__(self, token: str) -> None:
        super().__init__(
            base_url="https://api.github.com",
            rate=RateBudget(Rate.per_second(10), max_in_flight=50),
            auth=BearerAuth(token),
            http=HttpOptions(headers={"Accept": "application/vnd.github+json"}),
        )

    async def repository(self, owner: str, name: str) -> Repository | None:
        return await self.get(t"repos/{owner}/{name}", expect=REPOSITORY_OR_NONE)

    def open_issues(self, owner: str, name: str):
        return self.paginate(t"repos/{owner}/{name}/issues", expect=decoder(list[Issue]), pages=ISSUE_PAGES)

async with GitHub(token) as github:
    repository = await github.repository("python", "cpython")
    async for issue in github.open_issues("python", "cpython"):
        print(issue.number, issue.title)

Retries, pacing under GitHub's limits, pausing when its hourly limit is spent, following Link headers, and keeping the token on api.github.com all come from the base class.

Where to next¶

• Compared with others: what headgate does that tenacity, stamina or a hand-rolled httpx client does not, and when to pick those instead.
• Configuration: every option, and where it lives.
• Declared responses and errors: what an endpoint answers, and what it raises.
• Testing a client: fake networks, instant retries and failure schedules, with nothing private patched.