Possibility to add new transports at runtime

[noudin-ledger]

When I write a reusable client, I don't necessarily know all URLs and the transports that will be needed at client creation time.
It would be great to be able to mount new transports the same way it was possible in requests.session
This is a feature I'd be happy to develop if this is something you want.

[tomchristie]

Hi!

Nope, we don't support that.
We've got an interface for mounting multiple transports, but it's only exposed at client init. httpx.Client(mounts=...).
I don't see us changing that because it provides a clearer constraint around the lifespan of the connection pools used by the client.

If you want a dynamic approach, then you could implement this using a custom transport...

# Setup the initial transport config
dynamic_mountpoints = MyCustomTransport()
dynamic_mountpoints['https'] = httpx.HTTPTransport()

# Create the client
client = httpx.Client(transport=dynamic_mountpoints)

# Change the config dynamically
dynamic_mountpoints['https://www.example.com'] = ...

What kind of use-case are you trying to support here?

[noudin-ledger]

Well, I'm creating a client and the transport specifies the max retries of a request.
And I'd like to be able to have a different value for different endpoints.
But maybe having the dynamic mountpoints could do it?

[karpetrosyan]

I believe mounting transports is what you need.

Example:

import httpx

mounts = {
        "http://example.com": httpx.HTTPTransport(retries=5),
        "all://*:8000": httpx.HTTPTransport(retries=2),
}
client = httpx.Client(mounts=mounts)

[noudin-ledger]

This works if I know the URL when I create the client, but in case I want to have a reusable client, this would fail.

[karpetrosyan]

Here's an example of a custom transport that handles such cases, allowing you to dynamically add a new mount point in the proper manner.

import httpx
from httpx._utils import URLPattern


class DynamicMountpointsTransports(httpx.BaseTransport):
    def __init__(self, transport: httpx.BaseTransport | None = None):
        self._mounts: dict[URLPattern, httpx.BaseTransport | None] = {}
        self._defult_transport = transport or httpx.HTTPTransport()

    def handle_request(self, request: httpx.Request) -> httpx.Response:
        transport = self._transport_for_url(request.url)
        return transport.handle_request(request)

    def _transport_for_url(self, url: httpx.URL) -> httpx.BaseTransport:
        """
        Returns the transport instance that should be used for a given URL.
        This will either be the standard connection pool, or a proxy.
        """
        for pattern, transport in self._mounts.items():
            if pattern.matches(url):
                return self._defult_transport if transport is None else transport

        return self._defult_transport

    def __setitem__(self, pattern: str, transport: httpx.BaseTransport | None):
        self._mounts[URLPattern(pattern)] = transport
        self._mounts = dict(sorted(self._mounts.items()))

    def __delitem__(self, pattern: str):
        del self._mounts[URLPattern(pattern)]
        self._mounts = dict(sorted(self._mounts.items()))


dynamic_mountpoints = DynamicMountpointsTransports()
dynamic_mountpoints["all://"] = httpx.HTTPTransport(retries=5)
dynamic_mountpoints["https://google.com"] = httpx.HTTPTransport(retries=2)

client = httpx.Client(transport=dynamic_mountpoints)

[tomchristie]

Be careful with that private import.