TopolearnReal APIs need token refresh and request signing, and services need consistent logging of every call.
Simple auth works like Requests: auth=("user", "pass") for Basic, httpx.DigestAuth(...) for Digest. For anything more, subclass httpx.Auth and implement auth_flow(self, request) as a generator:
- modify the request (add a header) and
yieldit, - the yield expression returns the response,
- optionally inspect it and yield another request (for example after refreshing a token on a 401).
Because the flow is a generator, the same class works for both Client and AsyncClient, as long as it does no I/O of its own. If the flow must make its own HTTP calls (fetching a token), implement sync_auth_flow and async_auth_flow separately. Set requires_response_body = True if the flow reads the response body, or requires_request_body = True for signing schemes that hash the body.
Event hooks run on every request or response of a client:
httpx.Client(event_hooks={"request": [log_request], "response": [log_response]})Request hooks receive the Request just before sending; response hooks receive the Response after headers arrive, before the body is read, so call response.read() inside the hook if you need the body. A response hook calling response.raise_for_status() makes every error status raise for that client. With AsyncClient, hooks must be async functions.
Example
import httpx
class RefreshingToken(httpx.Auth):
def __init__(self):
self.token = "expired"
def auth_flow(self, request):
request.headers["Authorization"] = f"Bearer {self.token}"
response = yield request # send and get the response back
if response.status_code == 401:
self.token = "fresh" # pretend we refreshed it
request.headers["Authorization"] = f"Bearer {self.token}"
yield request # retry once
def handler(request):
ok = request.headers["Authorization"] == "Bearer fresh"
return httpx.Response(200 if ok else 401, json={"ok": ok})
log = []
client = httpx.Client(
transport=httpx.MockTransport(handler), auth=RefreshingToken(),
event_hooks={"response": [lambda r: log.append(r.status_code)]})
print(client.get("https://api.test/me").json()) # {'ok': True}
print(log) # [401, 200]Remember
- Subclass httpx.Auth and write auth_flow as a generator: yield request, receive response
- The same flow works sync and async if it does no I/O
- event_hooks run on every request/response; async clients need async hooks
Watch out for
- Reading response.json() in a response hook without calling response.read() first.