Recognizing httpx.Client settings with custom Transports

[lu-pl]

Hello,

I have a question regarding the behavior of httpx.Client in the context of custom Transports (i.e. httpx.BaseTransport subclasses).

I am implementing a SPARQL Protocol client that, apart from targeting remote endpoints, also supports running queries against in-memory Graph objects. The Graph object feature is implemented using custom Transport classes that essentially delegate any SPARQL operation request to the given Graph object instead of a remote endpoint.

These are my current Transport definitions: https://github.com/lu-pl/sparqlx/blob/main/src/sparqlx/utils/transports.py

I just realized that for any httpx.Client initialized with those Transports, client settings like e.g. timeouts are not recognized. It seems like httpx-defined transports like e.g. httpx.WSGITransport do not recognize client settings either (as far as I can tell).

So I am wondering: Are client configs even meaningful with custom transports? If yes, how to best implement client config recognition with custom transports?

Any help is much appreciated! :)

[murriiii]

Good question. Short answer: client settings like timeouts are not automatically forwarded to custom transports — this is by design.

The httpx.Client applies timeouts at the transport level for the built-in httpcore transport, but custom BaseTransport subclasses receive a raw Request object and are responsible for their own timeout handling.

Why this makes sense for your case

For your SPARQL in-memory graph transport, timeouts don't really apply — querying an in-memory Graph object is synchronous and effectively instant. So not having timeout support is fine.

However, if you wanted to respect client settings in a custom transport (e.g., for consistency or for transports that do actual I/O), here's how:

Accessing client config in a custom transport

The handle_request method receives Request which has an extensions dict. httpx passes timeout info via extensions:

import httpx

class GraphTransport(httpx.BaseTransport):
    def __init__(self, graph):
        self.graph = graph

    def handle_request(self, request: httpx.Request) -> httpx.Response:
        # Timeout is available in extensions (if you need it)
        timeout = request.extensions.get("timeout", {})
        # timeout is a dict like {"connect": 5.0, "read": 5.0, "write": 5.0, "pool": 5.0}

        # Your in-memory query logic
        result = self._query_graph(request)
        return httpx.Response(200, content=result)

So timeouts are passed through — you just need to check request.extensions["timeout"] and implement the timeout logic yourself if needed.

Other client settings

For other settings like base_url, headers, cookies, etc. — these are applied by the Client before calling the transport. So your transport's handle_request will already receive a request with:

• Fully resolved URL (including base_url)
• Merged headers (client defaults + per-request)
• Cookies applied

So those settings do work with custom transports — only timeout enforcement is left to the transport implementation.

Summary

Setting	Applied to custom transports?
base_url	Yes (URL resolved before transport)
headers	Yes (merged into Request)
cookies	Yes (merged into Request)
params	Yes (merged into URL)
timeout	Passed via request.extensions, but not enforced
max_redirects	No (handled by Client wrapper, not transport)
follow_redirects	No (handled by Client wrapper)

[vynazevedo]

One correction to the table above: follow_redirects and max_redirects still work with a custom transport. The client handles them around the transport call.

Here's a small example that runs on HTTPX 0.28.1, without making any network requests:

import httpx

class Transport(httpx.BaseTransport):
    def handle_request(self, request):
        assert request.extensions["timeout"]["read"] == 2.0
        if request.url.path == "/start":
            return httpx.Response(302, headers={"location": "/done"})
        return httpx.Response(200, text="ok")

with httpx.Client(
    transport=Transport(),
    base_url="https://example.test",
    timeout=2.0,
    follow_redirects=True,
) as client:
    response = client.get("/start")
    assert response.text == "ok"
    assert len(response.history) == 1

Setting max_redirects=0 on that client raises httpx.TooManyRedirects, also checked locally.

For your graph transport, the timeout values reach handle_request through request.extensions; HTTPX doesn't interrupt the graph query when that time expires. The graph/query engine would need to enforce a deadline or support cancellation. Being in memory doesn't necessarily make a SPARQL query fast, so I wouldn't rule out needing that limit.

The timeout extension docs describe the values passed to the transport. Headers, URL parameters and the base URL are already applied by the client before it calls you.

[Helloworldceo]

Adding a different approach, since the replies above show the timeout reaches your transport but isn't enforced. You can enforce it yourself: run the query in a worker thread and wait for it with the client's read timeout. Then client-level and per-request timeouts behave the same as with a network transport, and callers can catch httpx.ReadTimeout as usual.

import concurrent.futures

import httpx


class GraphTransport(httpx.BaseTransport):
    def __init__(self, run_query, max_workers: int = 4):
        self._run_query = run_query  # e.g. a function that queries your Graph and returns bytes
        self._pool = concurrent.futures.ThreadPoolExecutor(max_workers=max_workers)

    def handle_request(self, request: httpx.Request) -> httpx.Response:
        read_timeout = request.extensions.get("timeout", {}).get("read")  # None = no limit
        future = self._pool.submit(self._run_query, request)
        try:
            body = future.result(timeout=read_timeout)
        except concurrent.futures.TimeoutError:
            future.cancel()
            raise httpx.ReadTimeout("graph query exceeded the read timeout", request=request) from None
        return httpx.Response(
            200,
            content=body,
            headers={"content-type": "application/sparql-results+json"},
        )

    def close(self) -> None:
        self._pool.shutdown(wait=False, cancel_futures=True)

with httpx.Client(transport=GraphTransport(run_query), base_url="http://graph.local", timeout=0.2) as client:
    client.get("/sparql")               # raises httpx.ReadTimeout if the query takes > 0.2 s
    client.get("/sparql", timeout=5.0)  # per-request override works too

For AsyncClient, the same idea with AsyncBaseTransport:

import asyncio


class AsyncGraphTransport(httpx.AsyncBaseTransport):
    def __init__(self, run_query):
        self._run_query = run_query

    async def handle_async_request(self, request: httpx.Request) -> httpx.Response:
        read_timeout = request.extensions.get("timeout", {}).get("read")
        try:
            body = await asyncio.wait_for(asyncio.to_thread(self._run_query, request), read_timeout)
        except asyncio.TimeoutError:
            raise httpx.ReadTimeout("graph query exceeded the read timeout", request=request) from None
        return httpx.Response(200, content=body)

One limit: Python can't kill a thread, so a timed-out query keeps running in the background until it finishes. The caller just stops waiting. If you need a hard stop, run the query in a subprocess, or use a timeout in your query engine if it has one.

I ran both versions against httpx2 (same transport API as httpx 0.28) with a fast and a slow (1 s) query function: the slow one raises ReadTimeout after the client's timeout, and a per-request timeout= overrides it.