supporting http+unix URLs across many paths with a single Connector

[minrk]

I'm working on an application (JupyterHub) and adding support for unix domain sockets. In this situation, I will connect to many different domain sockets, while preserving the concurrent request limits of Connector, etc. so I don't want to specify a single socket as the UnixConnector requires, I want the path to come from the URL.

I want to support this URL scheme:

http+unix://{quote('socket/path', safe='')}/...

Two questions:

1. is there room for support of http+unix like this by default in aiohttp, or some other mechanism for resolving the path from the url? (fine if no, given the lack of standardization on where/how paths ought to be specified in http-over-unix). Even if UnixConnector had an overrideable path_from_url hook I could use in a subclass, that would be nice.
2. when writing a custom Connector to support this, what is the best/simplest method to override?

I currently have:

class _HTTPUnixConnector(TCPConnector):
    allowed_protocol_schema_set = {"http", "https", "http+unix"}
    
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._unix_connectors = WeakValueDictionary()
    
    def _get_connector(self, req):
        if req.url.scheme == "http+unix":
            path = unquote(req.url.host)
            connector = self._unix_connectors.get(path, None)
            if connector and not connector.closed:
                return connector
            connector = self._unix_connectors[path] = UnixConnector(path=path, limit=self.limit_per_host or self.limit)
            return connector
        else:
            return super()
    
    async def _create_connection(
            self, req: ClientRequest, traces: list, timeout: "ClientTimeout"
        ):
        return await self._get_connector(req)._create_connection(req, traces, timeout)

Is it safe to override the private (but documented) _create_connection method in a subclass outside aiohttp itself?

Is it better to use UnixConnector instances like this, or just copy the body of UnixConnector._create_connection and lose the multiple Connector instances to keep track of?

[Dreamsorcerer]

Is it safe to override the private (but documented) _create_connection method in a subclass outside aiohttp itself?

I think that's the expected approach, yes.

If you want to compare and contrast, here's an untested solution from Claude, which doesn't use nested connectors:

AI example

"""Custom connector that adds ``http+unix://`` URLs on top of plain TCP.

aiohttp ships :class:`aiohttp.UnixConnector`, but it is bound to a *single*
socket path given to its constructor and cannot also serve ordinary
``http://`` / ``https://`` requests. This example shows a connector that does
both: regular URLs go over TCP (handled by :class:`aiohttp.TCPConnector`, the
base class), while ``http+unix://`` URLs are routed to the Unix socket named in
the URL, so one session can reach many sockets *and* the network.

For ``http+unix://`` the socket path is carried in the URL host, percent-encoded
so its slashes survive URL parsing (the same convention used by
``requests-unixsocket`` and ``httpx``)::

    http+unix://%2Frun%2Fdocker.sock/v1.41/info
    http+unix://%2Ftmp%2Fapp.sock/status

Use :func:`unix_url` to build such URLs from an ordinary filesystem path.

Run it directly to drive a small server bound to both a Unix socket and a TCP
port::

    python examples/client_unix_socket.py
"""

import asyncio
import tempfile
from pathlib import Path
from urllib.parse import quote, unquote

from yarl import URL

from aiohttp import (
    ClientConnectorError,
    ClientRequest,
    ClientSession,
    ClientTimeout,
    TCPConnector,
    web,
)
from aiohttp.client_proto import ResponseHandler
from aiohttp.tracing import Trace


class UnixURLConnector(TCPConnector):
    """Route ``http+unix://`` URLs to Unix sockets, everything else over TCP.

    Subclassing :class:`~aiohttp.TCPConnector` means ``http``/``https`` (and
    ``ws``/``wss``) keep working exactly as usual; only the extra
    ``http+unix://`` scheme is handled specially. Unlike
    :class:`aiohttp.UnixConnector`, the socket path is not fixed at construction
    time: it is read from the (percent-encoded) host of each ``http+unix://``
    URL, so a single session can reach many sockets.
    """

    # Advertise the extra scheme on top of the ones TCPConnector accepts,
    # otherwise ClientSession rejects http+unix:// URLs.
    allowed_protocol_schema_set = TCPConnector.allowed_protocol_schema_set | frozenset(
        {"http+unix"}
    )

    async def _create_connection(
        self,
        req: ClientRequest,
        traces: list[Trace],
        timeout: ClientTimeout,
    ) -> ResponseHandler:
        if req.url.scheme != "http+unix":
            # Ordinary URL: let TCPConnector do its normal thing.
            return await super()._create_connection(req, traces, timeout)

        # req.url.host is the percent-encoded socket path; decode it back into
        # a filesystem path such as "/run/docker.sock".
        path = unquote(req.url.host or "")
        try:
            _, proto = await asyncio.wait_for(
                # _loop and _factory are provided by BaseConnector and are what
                # every built-in connector uses to build the protocol.
                self._loop.create_unix_connection(self._factory, path),
                timeout.sock_connect,
            )
        except asyncio.TimeoutError:
            # Let a connect timeout propagate; only genuine connection errors
            # become ClientConnectorError (TimeoutError is an OSError on 3.11+).
            raise
        except OSError as exc:
            raise ClientConnectorError(req.connection_key, exc) from exc
        return proto


def unix_url(socket_path: str, path: str = "/") -> URL:
    """Build an ``http+unix://`` URL targeting *socket_path*.

    The socket path is percent-encoded into the URL host so that an ordinary
    request path (and query string) can follow it.
    """
    return URL(f"http+unix://{quote(socket_path, safe='')}").with_path(path)


async def handle(request: web.Request) -> web.Response:
    return web.json_response(
        {"message": f"hello from {request.host}", "path": request.path}
    )


async def main() -> None:
    with tempfile.TemporaryDirectory() as tmp:
        socket_path = str(Path(tmp) / "demo.sock")

        # One app served on both a Unix socket and a TCP port.
        app = web.Application()
        app.router.add_get("/greet", handle)
        runner = web.AppRunner(app)
        await runner.setup()
        await web.UnixSite(runner, socket_path).start()
        await web.TCPSite(runner, "127.0.0.1", 0).start()
        host, port = next(a for a in runner.addresses if isinstance(a, tuple))[:2]

        try:
            async with ClientSession(connector=UnixURLConnector()) as session:
                # http+unix:// -> the Unix socket named in the URL.
                async with session.get(unix_url(socket_path, "/greet")) as resp:
                    print("GET", resp.url)
                    print(" ->", resp.status, await resp.json())

                # http:// -> ordinary TCP, handled by the TCPConnector base.
                async with session.get(f"http://{host}:{port}/greet") as resp:
                    print("GET", resp.url)
                    print(" ->", resp.status, await resp.json())

                # A missing socket surfaces as a normal ClientConnectorError.
                try:
                    await session.get(unix_url("/tmp/missing.sock", "/greet"))
                except ClientConnectorError as exc:
                    print("expected failure:", exc)
        finally:
            await runner.cleanup()


if __name__ == "__main__":
    asyncio.run(main())