Client integrations

proxylib.integrations.requests

requests integration.

ProxyMapAdapter (requires the requests package -- pip install requests; there is no proxylib extra for it) is the recommended way to wire a ProxyMap into a requests.Session — it hooks HTTPAdapter.send(), so it sees the real request URL. For the simpler proxies-dict style (requests.get(url, proxies=...)) use :class:proxylib.integrations.dict.ProxyDict, which isn't requests-specific.

__all__ = ('ProxyMapAdapter',) module-attribute

__getattr__(name)

Source code in src/proxylib/integrations/requests.py

def __getattr__(name: str) -> Any:
    if name == "ProxyMapAdapter":
        cls = _build_adapter_class()
        # Only the real class is kept: requests may be importable later (it
        # can be mid-import now).
        if not getattr(cls, "_placeholder", False):
            globals()["ProxyMapAdapter"] = cls
        return cls
    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")

proxylib.integrations.urllib

urllib.request integration.

ProxyMapHandler resolves the proxy per-request from a ProxyMap, rather than a static {scheme: proxy_url} dict like the stdlib urllib.request.ProxyHandler normally uses -- the same upgrade :class:proxylib.integrations.requests.ProxyMapAdapter is over a plain proxies dict, and for the same reason: a static dict can't see the request's host/path, which most ProxyMap rules (PAC scripts, path-dependent NO_PROXY) need.

__all__ = ('ProxyMapHandler',) module-attribute

ProxyMapHandler(proxymap)

Bases: ProxyHandler

A urllib.request handler that resolves proxies per-request from a ProxyMap::

opener = urllib.request.build_opener(ProxyMapHandler(proxymap))
opener.open("https://example.com")

# or, to affect every urllib.request.urlopen() call:
urllib.request.install_opener(opener)

When the map has no opinion (KeyError), the request uses the proxy urllib.request.getproxies() names for its scheme, as the standard ProxyHandler would -- read by proxylib when it is a PAC result or a gateway, which urllib cannot use itself.

A URL-prefix gateway gets the request at <gateway><URL> through the opener's own HTTP(S) handler, and the response keeps the target URL, so the cookie processor and the redirect handler see the target.

urllib speaks HTTP proxies only, and sends a proxy's headers through a tunnel unless they are Proxy-Authorization: SOCKS proxies, and (for an https URL) proxies with other headers, are skipped, and an answer with nothing else raises :class:~proxylib.UnsupportedProxyError.

Source code in src/proxylib/integrations/urllib.py

def __init__(self, proxymap: ProxyMap) -> None:
    # proxies={}: skip ProxyHandler.__init__'s per-protocol setattr(), which
    # would otherwise shadow the http_open/https_open/ftp_open methods
    # below with static, env-derived ones (its default when proxies=None).
    super().__init__(proxies={})
    self.proxymap = proxymap

ftp_open = _resolve class-attribute instance-attribute

http_open = _resolve class-attribute instance-attribute

https_open = _resolve class-attribute instance-attribute

proxymap = proxymap instance-attribute

proxylib.integrations.httpx

httpx integration (needs httpx; there is no proxylib extra for it).

ProxyTransport and AsyncProxyTransport resolve the proxy for every request from a proxy map and fail over along its answer. httpx has no global hook, so a client opts in::

client = httpx.Client(transport=ProxyTransport(proxymap))
client = httpx.AsyncClient(transport=AsyncProxyTransport(proxymap))

__all__ = ('AsyncProxyTransport', 'ProxyTransport') module-attribute

__getattr__(name)

Source code in src/proxylib/integrations/httpx.py

def __getattr__(name: str) -> Any:
    if name in __all__:
        classes = _build()
        if not getattr(classes[name], "_placeholder", False):
            globals().update(classes)
        return classes[name]
    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")

proxylib.integrations.aiohttp

aiohttp integration (needs aiohttp; there is no proxylib extra for it).

aiohttp has no transport to plug a resolver into, so :class:ProxySession wraps a ClientSession: every request asks the proxy map and is sent with proxy= set to the chosen entry, failing over along the answer::

async with ProxySession(proxymap) as session:
    async with session.get("https://example.com") as response:
        body = await response.text()

__all__ = ('ProxySession',) module-attribute

ProxySession(proxymap, session=None, *, failover=True, bad_proxy_ttl=300.0, **session_kwargs)

An aiohttp.ClientSession whose requests take their proxy from proxymap (read with :func:~proxylib.alookup).

session is the session to send through; by default one is made, with session_kwargs, on the first request and closed by :meth:close. A session passed in is not closed. request(method, url, **kwargs) and get/post/put/patch/delete/head/options take ClientSession.request's arguments.

The answer's entries are tried in order, healthy ones first; with failover (the default), one that fails to connect (aiohttp.ClientConnectorError, which includes ClientProxyConnectionError, or a connect timeout) is marked bad for bad_proxy_ttl seconds and the next is tried, and the last error is raised. When the map has no opinion the request goes as aiohttp would send it. A proxy= argument that is a proxylib value (a PAC result such as "PROXY a:1; DIRECT", a list of proxies, a Proxy) is resolved and failed over instead of the map; a plain proxy URL is aiohttp's own.

A proxy's headers go to that proxy only: in the CONNECT of an https request, as headers of a plain-http one (sent to the proxy). aiohttp speaks HTTP proxies only: SOCKS proxies and URL-prefix gateways in an answer are skipped, and an answer with nothing else raises :class:~proxylib.UnsupportedProxyError. Unless the aiohttp patch target is installed (the default), a session made with trust_env=True uses the environment's proxy for a DIRECT entry, so leave that off when the map decides.

Source code in src/proxylib/integrations/aiohttp.py

def __init__(
    self,
    proxymap: ProxyMap,
    session: Any = None,
    *,
    failover: bool = True,
    bad_proxy_ttl: float = 300.0,
    **session_kwargs: Any,
) -> None:
    _aiohttp()
    if session is not None and session_kwargs:
        raise TypeError("pass either a session or ClientSession arguments, not both")
    self.proxymap = proxymap
    self.failover = failover
    self.bad_proxy_ttl = bad_proxy_ttl
    self._session = session
    self._owns = session is None
    self._session_kwargs = session_kwargs

bad_proxy_ttl = bad_proxy_ttl instance-attribute

failover = failover instance-attribute

proxymap = proxymap instance-attribute

session property

The aiohttp.ClientSession, made on first use.

__aenter__() async

Source code in src/proxylib/integrations/aiohttp.py

async def __aenter__(self) -> "ProxySession":
    return self

__aexit__(*exc_info) async

Source code in src/proxylib/integrations/aiohttp.py

async def __aexit__(self, *exc_info: object) -> None:
    await self.close()

close() async

Close the session this wrapper made; a session passed in stays open.

Source code in src/proxylib/integrations/aiohttp.py

async def close(self) -> None:
    """Close the session this wrapper made; a session passed in stays open."""
    if self._owns and self._session is not None:
        await self._session.close()

delete(url, **kwargs)

Source code in src/proxylib/integrations/aiohttp.py

def delete(self, url: Any, **kwargs: Any) -> _Request:
    return self.request("DELETE", url, **kwargs)

get(url, **kwargs)

Source code in src/proxylib/integrations/aiohttp.py

def get(self, url: Any, **kwargs: Any) -> _Request:
    return self.request("GET", url, **kwargs)

head(url, **kwargs)

Source code in src/proxylib/integrations/aiohttp.py

def head(self, url: Any, **kwargs: Any) -> _Request:
    return self.request("HEAD", url, **kwargs)

options(url, **kwargs)

Source code in src/proxylib/integrations/aiohttp.py

def options(self, url: Any, **kwargs: Any) -> _Request:
    return self.request("OPTIONS", url, **kwargs)

patch(url, **kwargs)

Source code in src/proxylib/integrations/aiohttp.py

def patch(self, url: Any, **kwargs: Any) -> _Request:
    return self.request("PATCH", url, **kwargs)

post(url, **kwargs)

Source code in src/proxylib/integrations/aiohttp.py

def post(self, url: Any, **kwargs: Any) -> _Request:
    return self.request("POST", url, **kwargs)

put(url, **kwargs)

Source code in src/proxylib/integrations/aiohttp.py

def put(self, url: Any, **kwargs: Any) -> _Request:
    return self.request("PUT", url, **kwargs)

request(method, url, **kwargs)

Source code in src/proxylib/integrations/aiohttp.py

def request(self, method: str, url: Any, **kwargs: Any) -> _Request:
    return _Request(self._request(method, url, **kwargs))

proxylib.integrations.dict

Plain proxies-dict integration: ProxyDict duck-types the {scheme_or_url: "proxy://uri"} mapping requests and similar libraries accept, resolving from a ProxyMap.

__all__ = ('ProxyDict',) module-attribute

ProxyDict(proxymap)

Duck-types the plain {scheme_or_url: "proxy://uri"} proxies dict that requests and similar libraries accept, resolving from a ProxyMap.

The simpler integration when a Transport Adapter/handler isn't wanted::

requests.get(url, proxies=ProxyDict(proxymap))

Consumers of a proxies dict look keys up by scheme/host, not the full request URL, so path-dependent rules (most PAC scripts) won't apply -- use ProxyMapAdapter/ProxyMapHandler when that fidelity matters.

Lookups return the first proxy's URI string; DIRECT raises KeyError (a missing key means "no proxy", matching the dict convention). A URL string cannot name a URL-prefix gateway or carry proxy headers: such entries are skipped, and an answer with nothing else raises :class:~proxylib.UnsupportedProxyError. ProxyMapAdapter resolves a ProxyDict itself and uses those entries.

Deliberately composes a ProxyMap rather than subclassing it -- __getitem__ here returns a str (a proxy URI), not Iterable[Optional[Proxy]], which would be an LSP violation if this were treated as one.

Source code in src/proxylib/integrations/dict.py

def __init__(self, proxymap: ProxyMap) -> None:
    self.proxymap = proxymap

__slots__ = ('proxymap',) class-attribute instance-attribute

proxymap = proxymap instance-attribute

__contains__(key)

Source code in src/proxylib/integrations/dict.py

def __contains__(self, key: object) -> bool:
    try:
        self[key]
        return True
    except KeyError:
        return False

__getitem__(uri)

Source code in src/proxylib/integrations/dict.py

def __getitem__(self, uri: str) -> str:
    # requests also asks for bare keys such as "http" or "all": a key
    # that is not a URL with a scheme and a host has no entry.
    parts = urlsplit(uri) if isinstance(uri, str) else None
    if parts is None or not parts.scheme or not parts.netloc:
        raise KeyError(uri)
    try:
        entries = lookup(self.proxymap, uri)
    except ValueError:
        raise KeyError(uri) from None
    proxy = usable(entries, "a proxies dict", uri, proxy_url_reason)[0]
    if proxy is None:
        raise KeyError(uri)
    return proxy.uri()

copy()

Source code in src/proxylib/integrations/dict.py

def copy(self) -> "ProxyDict":
    return type(self)(self.proxymap)

get(uri, default=None)

Source code in src/proxylib/integrations/dict.py

def get(self, uri: str, default=None):
    try:
        return self[uri]
    except KeyError:
        return default

setdefault(url, value)

No-op: requests calls this to merge env proxies; the ProxyMap wins.

Source code in src/proxylib/integrations/dict.py

def setdefault(self, url: str, value: str) -> None:
    """No-op: requests calls this to merge env proxies; the ProxyMap wins."""