TopolearnTransports are the cleanest extension point in HTTPX for cross-cutting behaviour like status retries.
A transport is any object implementing handle_request(self, request) -> Response (sync, httpx.BaseTransport) or async handle_async_request(self, request) (async, httpx.AsyncBaseTransport), plus optional close() / aclose(). Everything the client does, such as auth, redirects, cookies and hooks, happens above the transport, which only moves a single request to a single response.
The usual pattern is a wrapping transport that delegates to the real one:
class RetryTransport(httpx.BaseTransport):
def __init__(self, inner): self.inner = inner
def handle_request(self, request):
... # call self.inner.handle_request(request) in a loopUses include retrying 429 and 503 responses with backoff, caching (hishel is built this way), request signing at the lowest level, metrics, and fault injection.
Details that matter:
- The response from the inner transport has an unread stream. If you discard it (to retry), call
response.close()first so the connection returns to the pool. If you need the body, callresponse.read(). - Request bodies that are generators can only be sent once; retrying requires a re-readable body (bytes).
- Transports see each hop of a redirect separately, because redirects are handled by the client.
- Mount different transports per URL pattern with
mounts=to apply behaviour only to some hosts.
Transports keep these concerns out of business code and out of auth or hook logic.
Example
import time
import httpx
class RetryTransport(httpx.BaseTransport):
def __init__(self, inner, retries=3, statuses=(429, 503)):
self.inner, self.retries, self.statuses = inner, retries, statuses
def handle_request(self, request):
for attempt in range(self.retries + 1):
response = self.inner.handle_request(request)
if response.status_code not in self.statuses or attempt == self.retries:
return response
response.close() # release the connection
time.sleep(0.05 * 2 ** attempt) # exponential backoff
def close(self):
self.inner.close()
attempts = []
def flaky(request):
attempts.append(1)
return httpx.Response(503 if len(attempts) < 3 else 200, text="done")
with httpx.Client(transport=RetryTransport(httpx.MockTransport(flaky))) as c:
r = c.get("https://api.test/job")
print(r.status_code, r.text, len(attempts)) # 200 done 3Remember
- A transport turns one Request into one Response; client features sit above it
- Wrap the default transport to add retries, caching or metrics
- Close discarded responses so connections return to the pool
Watch out for
- Retrying a request whose body was a one-shot generator.