Core

proxylib.proxy

Core proxy/URI types: parsing proxy strings and mapping request URLs to proxies.

Proxy models a single scheme://[user[:pass]@]host[:port] proxy entry (as found in PROXY/env-var/PAC proxy strings). ProxyMap is the protocol every proxy-selection strategy in this library implements (EnvProxyConfig, pac.PAC, SimpleProxyMap): given a request URL it returns the sequence of Proxy (or None for DIRECT) to try, in order.

__all__ = ['Proxy', 'ProxyMap', 'UriSplit', 'SimpleProxyMap', 'ChainProxyMap', 'ConfigurableProxyMap'] module-attribute

ChainProxyMap(*maps)

Bases: ProxyMap

Tries each ProxyMap in order; the first one with an opinion wins.

Works because of the codified result contract (see ProxyMap's docstring): a constituent map's KeyError means "no opinion, try the next one", while a [None]/[Proxy] result is definitive and stops the chain immediately -- e.g. ChainProxyMap(EnvProxyConfig(...), pac_map) falls through to the PAC map only when the env config has no proxy configured for that scheme, not merely when it resolves to DIRECT.

If every constituent map raises KeyError, the chain does too -- "no opinion" composes, so a ChainProxyMap can itself be one link in a larger chain.

Source code in src/proxylib/proxy.py

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

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

maps = maps instance-attribute

__getitem__(uri)

Source code in src/proxylib/proxy.py

def __getitem__(self, uri: str) -> Iterable[Optional[Proxy]]:
    for proxymap in self.maps:
        try:
            return proxymap[uri]
        except KeyError:
            continue
    raise KeyError(uri)

ConfigurableProxyMap(proxymap, *, cache_ttl=None, probe=False, probe_timeout=5.0, round_robin=False, browser_compatibility=False, bypass_local=False, no_proxy=None)

Bases: ProxyMap

Decorates any ProxyMap with caching, active probing, round-robin selection, browser-style privacy stripping, and an implicit local bypass.

All features are opt-in and independent; the decorated map's own result contract (KeyError/[None]/[Proxy]) is preserved throughout.

Source code in src/proxylib/proxy.py

def __init__(
    self,
    proxymap: ProxyMap,
    *,
    cache_ttl: "float|None" = None,
    probe: bool = False,
    probe_timeout: float = 5.0,
    round_robin: bool = False,
    browser_compatibility: bool = False,
    bypass_local: bool = False,
    no_proxy: "Iterable[str]|None" = None,
) -> None:
    self.proxymap = proxymap
    self.cache_ttl = cache_ttl
    self.probe = probe
    self.probe_timeout = probe_timeout
    self.round_robin = round_robin
    self.browser_compatibility = browser_compatibility

    # Local import: env.py imports proxy.py, so importing it back at
    # module level here would be circular (same reason ProxyMap.__new__
    # locally imports `pac`). Reused rather than reimplemented: an
    # EnvProxyConfig(None, None, rules) has no configured http/https
    # proxy, so it raises KeyError for anything NOT matched by `rules`
    # and returns [None] for anything that is -- exactly the "bypass
    # check" primitive this class needs, complete with CIDR/<local>
    # matching and the Phase 2 global no_proxy defaults, for free.
    bypass_rules = list(no_proxy or ())
    if bypass_local:
        # <local> already covers loopback + link-local (env.py shares
        # netutils.is_loopback_or_link_local for that) plus same-subnet
        # addresses -- a superset of what bypass_local asks for.
        bypass_rules.append("<local>")
    self._bypass_checker: "Optional[ProxyMap]" = None
    if bypass_rules:
        from .env import EnvProxyConfig

        self._bypass_checker = EnvProxyConfig(None, None, bypass_rules)

    self._cache: "Dict[str, Tuple[float, tuple]]" = {}
    self._round_robin_index = 0
    self._lock = threading.Lock()

browser_compatibility = browser_compatibility instance-attribute

cache_ttl = cache_ttl instance-attribute

probe = probe instance-attribute

probe_timeout = probe_timeout instance-attribute

proxymap = proxymap instance-attribute

round_robin = round_robin instance-attribute

__getitem__(url)

Source code in src/proxylib/proxy.py

def __getitem__(self, url: str) -> Iterable[Optional[Proxy]]:
    if self._bypass_checker is not None:
        try:
            return self._bypass_checker[url]
        except KeyError:
            pass  # not bypassed -- fall through to normal resolution

    effective_url = self._effective_url(url)

    if self.cache_ttl:
        cached = self._cache.get(effective_url)
        if cached is not None and (time.monotonic() - cached[0]) < self.cache_ttl:
            return cached[1]

    result: tuple = tuple(self.proxymap[effective_url])

    if self.round_robin and len(result) > 1:
        with self._lock:
            index = self._round_robin_index % len(result)
            self._round_robin_index += 1
        result = result[index:] + result[:index]

    if self.probe:
        try:
            result = (first_working_proxy(result, timeout=self.probe_timeout),)
        except LookupError:
            # No candidate was reachable -- a resolution failure, not a
            # DIRECT decision (same reasoning as LibProxyMap/
            # CFNetworkProxyMap's timeout/failure handling).
            raise KeyError(url)

    if self.cache_ttl:
        self._cache[effective_url] = (time.monotonic(), result)

    return result

clear_cache()

Source code in src/proxylib/proxy.py

def clear_cache(self) -> None:
    self._cache.clear()

Proxy

Bases: _URI

A single proxy authority, e.g. http://user:pass@proxy.example.com:8080.

Proxy("direct", ...) returns None (the sentinel used everywhere in this library to mean "connect without a proxy") rather than an instance.

url property

__new__(scheme, username, password, host, port)

Source code in src/proxylib/proxy.py

def __new__(
    cls,
    scheme: str,
    username: "str|None",
    password: "str|None",
    host: "str|None",
    port: "str|int|None",
) -> "Optional[Proxy]":
    scheme = (scheme or "").lower()
    if scheme == "direct":
        return None
    elif scheme == "proxy":
        scheme = "http"
    elif scheme == "socks":
        scheme = "socks4"
    elif not scheme:
        scheme = cls._DEFAULT_SCHEME

    if port:
        port = int(port)

    return super().__new__(
        cls, scheme, username or "", password or "", host or "", port or 0
    )

ProxyMap

Bases: Protocol

Protocol for "given a request URL, which proxies should I try?".

Calling ProxyMap(src) is a small factory: a plain proxy-authority string builds a SimpleProxyMap, but a string that looks like a URL to a PAC file/script is loaded as one (see :mod:proxylib.pac).

Result contract (every implementation must follow this so chaining multiple maps together is meaningful):

• __getitem__ raises KeyError to mean no decision -- this map has no opinion on the request, distinct from an explicit answer. A fallback/chain should try the next map.
• __getitem__ yields None as an entry to mean DIRECT (no proxy) -- an explicit, definitive answer, not "no opinion".
• __getitem__ yields a Proxy entry to mean use this proxy.

Per-implementation notes: SimpleProxyMap and pac.PAC are always definitive (constructed with, or computed, an explicit answer) and never raise KeyError. EnvProxyConfig raises KeyError for a scheme with no configured env proxy (env vars can't express "explicitly DIRECT", only "set" or "absent").

__contains__(key)

Source code in src/proxylib/proxy.py

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

__enter__()

Source code in src/proxylib/proxy.py

def __enter__(self):
    # Local import: patching.py imports proxy.py, so a top-level import
    # here would be circular (same reason ProxyMap.__new__ locally
    # imports `pac`).
    from . import patching

    patching.patch(self)
    return self

__exit__(exc_type, exc_val, exc_tb)

Source code in src/proxylib/proxy.py

def __exit__(self, exc_type, exc_val, exc_tb) -> bool:
    from . import patching

    patching.unpatch()
    return False

__getitem__(uri)

Source code in src/proxylib/proxy.py

def __getitem__(self, uri: str) -> Iterable[Optional[Proxy]]:
    raise NotImplementedError()

__new__(*args, **kwargs)

Source code in src/proxylib/proxy.py

def __new__(cls, *args, **kwargs):
    src: "str|Proxy|None" = args[0] if args else None
    if cls is ProxyMap:
        if isinstance(src, str):
            looks_like_pac_source = _looks_like_pac_source(src)
            if looks_like_pac_source:
                from . import pac

                return pac.load(src)
        return object.__new__(SimpleProxyMap)

    return object.__new__(cls)

get(uri, default=None)

Source code in src/proxylib/proxy.py

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

SimpleProxyMap(proxy=None)

Bases: ProxyMap

A ProxyMap that always returns the same fixed proxy (or list of proxies).

Source code in src/proxylib/proxy.py

def __init__(self, proxy: "Proxy|Sequence[Optional[Proxy]]|str|None" = None) -> None:
    if isinstance(proxy, str):
        # PAC-style strings ("DIRECT", "PROXY host:port; DIRECT") have no
        # "://"; URL-style strings ("http://host:port", the conventional
        # HTTP_PROXY/HTTPS_PROXY env var format) do.
        fmt = UriSplit.Default if "://" in proxy else UriSplit.PAC
        proxy = Proxy.find_all(proxy, fmt)
    if proxy is None or isinstance(proxy, Proxy):
        self.proxies: Sequence[Optional[Proxy]] = (proxy,)
    elif isinstance(proxy, typing.Sequence):
        self.proxies = proxy
    else:
        self.proxies = (proxy,)

proxies = (proxy,) instance-attribute

__getitem__(uri)

Source code in src/proxylib/proxy.py

def __getitem__(self, uri: str) -> Sequence[Optional[Proxy]]:
    return self.proxies

UriSplit

Bases: Enum

Regexes for splitting either a plain URI or a PAC PROXY ...; ... string.

Default = re.compile(f'{DELIM}(?:(?:({SCHEME}):)?(?://{AUTHORITY})?\s*)') class-attribute instance-attribute

PAC = re.compile(f'{DELIM}({SCHEME})(?:\s+(?:{AUTHORITY})?\s*)?') class-attribute instance-attribute

findall(uri)

Source code in src/proxylib/_uri.py

def findall(self, uri: str):
    return self.value.findall(uri)

match(uri)

Source code in src/proxylib/_uri.py

def match(self, uri: str):
    return self.value.match(uri)

proxylib.env

Proxy selection from the conventional HTTP_PROXY/HTTPS_PROXY/NO_PROXY env vars.

__all__ = ('EnvProxyConfig', 'set_default_no_proxy', 'get_default_no_proxy') module-attribute

EnvProxyConfig(http_proxy, https_proxy, no_proxy)

Bases: ProxyMap

A ProxyMap built from HTTP_PROXY/HTTPS_PROXY/NO_PROXY-style settings.

Source code in src/proxylib/env.py

def __init__(
    self,
    http_proxy: "str|Proxy|None",
    https_proxy: "str|Proxy|None",
    no_proxy: "Iterable[str]|None",
) -> None:
    # SimpleProxyMap directly, not the ProxyMap factory: proxy env var
    # values are never PAC sources (PROXY_PAC/AutoConfigURL cover that
    # separately in each OS backend), and the factory routing a
    # PAC-looking value to pac.load() would do network I/O from this
    # constructor.
    # None means "no opinion" (KeyError), not "explicitly DIRECT" -- env
    # vars have no way to express DIRECT, only "set" or "absent". Keeping
    # these Optional (instead of always building a SimpleProxyMap) lets
    # __getitem__ raise KeyError for an unconfigured scheme so a
    # ChainProxyMap can fall through to the next map instead of the
    # (wrong) DIRECT result SimpleProxyMap(None) would otherwise give.
    self.http_proxy: "Optional[ProxyMap]" = SimpleProxyMap(http_proxy) if http_proxy else None
    self.https_proxy: "Optional[ProxyMap]" = SimpleProxyMap(https_proxy) if https_proxy else None
    # Explicit rules are deduped-and-checked ahead of the process-wide
    # defaults, but since NO_PROXY rules are purely additive (matching
    # any entry bypasses the proxy) there's no override semantics to get
    # wrong here -- this just controls iteration order.
    merged = dict.fromkeys([*(no_proxy or ()), *get_default_no_proxy()])
    self.no_proxy: "List[_NoProxyEntry]" = [_parse_no_proxy_entry(_no) for _no in merged]

__slots__ = ('http_proxy', 'https_proxy', 'no_proxy') class-attribute instance-attribute

http_proxy = SimpleProxyMap(http_proxy) if http_proxy else None instance-attribute

https_proxy = SimpleProxyMap(https_proxy) if https_proxy else None instance-attribute

no_proxy = [(_parse_no_proxy_entry(_no)) for _no in merged] instance-attribute

__getitem__(url)

Source code in src/proxylib/env.py

def __getitem__(self, url: str) -> Iterable[Optional[Proxy]]:
    uri = URL.from_str(url)
    # Resolve DNS lazily: get_ip() is a blocking gethostbyname() call,
    # only needed for "<local>" entries -- don't pay it per lookup when
    # no_proxy has none (the overwhelmingly common case).
    ip_literal: "_ip.IPv4Address|_ip.IPv6Address|None" = None
    ip_literal_checked = False
    for entry in self.no_proxy:
        if entry is None:
            # "<local>": bypass the proxy for loopback/link-local addresses
            # (shared with ConfigurableProxyMap's bypass_local=) and for
            # addresses on the same subnet as a local interface.
            ip = get_ip(uri.host)
            if ip is None:
                continue
            if is_loopback_or_link_local(ip):
                return [None]
            for _if in get_local_interfaces():
                if ip in _if.network:
                    return [None]
        elif isinstance(entry, (_ip.IPv4Network, _ip.IPv6Network)):
            # CIDR entries match IP-literal request hosts only (curl
            # semantics) -- resolving hostnames to check membership
            # would reintroduce the per-lookup DNS cost this module
            # otherwise deliberately avoids (see the known-bugs note).
            if not ip_literal_checked:
                try:
                    ip_literal = _ip.ip_address(uri.host)
                except ValueError:
                    ip_literal = None
                ip_literal_checked = True
            if ip_literal is not None and ip_literal in entry:
                return [None]
        elif _no_proxy_matches(uri.host, uri.port, entry):
            return [None]
    target = self.https_proxy if uri.scheme == "https" else self.http_proxy
    if target is None:
        raise KeyError(url)
    return target[f"{uri.scheme}://{uri.netloc}"]

from_env() staticmethod

Build an EnvProxyConfig from the process environment (upper or lowercase names).

Source code in src/proxylib/env.py

@staticmethod
def from_env() -> "EnvProxyConfig":
    """Build an ``EnvProxyConfig`` from the process environment (upper or lowercase names)."""
    https = os.environ.get("HTTPS_PROXY") or os.environ.get("https_proxy")
    http = os.environ.get("HTTP_PROXY") or os.environ.get("http_proxy")
    no_proxy = os.environ.get("NO_PROXY") or os.environ.get("no_proxy")

    return EnvProxyConfig(
        http, https, [_no.strip() for _no in no_proxy.split(",")] if no_proxy else []
    )

get_default_no_proxy()

Return a copy of the current process-wide default NO_PROXY rules.

Source code in src/proxylib/env.py

def get_default_no_proxy() -> "List[str]":
    """Return a copy of the current process-wide default ``NO_PROXY`` rules."""
    return list(_default_no_proxy)

set_default_no_proxy(rules)

Set the process-wide default NO_PROXY rules (same entry syntax as the env var: hostnames, .suffix, <local>, CIDR). Merged with -- not replaced by -- whatever rules each EnvProxyConfig/ConfigurableProxyMap is given explicitly. Pass None (or an empty iterable) to clear it.

Source code in src/proxylib/env.py

def set_default_no_proxy(rules: "Iterable[str]|None") -> None:
    """Set the process-wide default ``NO_PROXY`` rules (same entry syntax as the
    env var: hostnames, ``.suffix``, ``<local>``, CIDR). Merged with -- not
    replaced by -- whatever rules each ``EnvProxyConfig``/``ConfigurableProxyMap``
    is given explicitly. Pass ``None`` (or an empty iterable) to clear it."""
    global _default_no_proxy
    _default_no_proxy = list(rules) if rules else []

proxylib.netutils

Small stdlib-first networking helpers used across proxylib.

get_local_interfaces prefers the optional ifaddr package (accurate adapter/prefix info on every OS) and falls back to a best-effort, stdlib-only implementation when it isn't installed.

clear_circuit_breaker()

Source code in src/proxylib/netutils.py

def clear_circuit_breaker() -> None:
    _circuit_breaker.clear()

clear_interfaces_cache()

Clear the get_local_interfaces cache. Call this in tests that monkeypatch enumeration.

Source code in src/proxylib/netutils.py

def clear_interfaces_cache() -> None:
    """Clear the ``get_local_interfaces`` cache. Call this in tests that monkeypatch enumeration."""
    global _interfaces_cache
    _interfaces_cache = None

first_working_proxy(proxies, timeout=5.0, circuit_breaker_ttl=30.0, clock=_time.monotonic)

Return the first entry that accepts a TCP connection, in order.

Failover helper for ProxyMap results (PROXY a; PROXY b; DIRECT means "try these in order")::

proxy = first_working_proxy(proxymap[url])

A None entry (DIRECT) is returned immediately -- there's no proxy to probe. This only checks TCP reachability of the proxy port, not that the proxy will actually serve the request. Raises LookupError when no entry is reachable (None can't signal failure here: it means DIRECT).

A proxy that fails its probe is blacklisted (skipped without probing) for circuit_breaker_ttl seconds (default 30); pass 0/None to disable.

Source code in src/proxylib/netutils.py

def first_working_proxy(
    proxies: "Iterable[Optional[Proxy]]",
    timeout: float = 5.0,
    circuit_breaker_ttl: "float|None" = 30.0,
    clock: "Callable[[], float]" = _time.monotonic,
) -> "Optional[Proxy]":
    """Return the first entry that accepts a TCP connection, in order.

    Failover helper for ``ProxyMap`` results (``PROXY a; PROXY b; DIRECT``
    means "try these in order")::

        proxy = first_working_proxy(proxymap[url])

    A ``None`` entry (DIRECT) is returned immediately -- there's no proxy to
    probe. This only checks TCP reachability of the proxy port, not that the
    proxy will actually serve the request. Raises ``LookupError`` when no
    entry is reachable (``None`` can't signal failure here: it means DIRECT).

    A proxy that fails its probe is blacklisted (skipped without probing)
    for ``circuit_breaker_ttl`` seconds (default 30); pass ``0``/``None`` to
    disable.
    """
    now = clock()
    for proxy in proxies:
        if proxy is None:
            return None
        port = proxy.port or get_default_port(proxy.scheme)
        if not port:
            continue
        key = (proxy.host, port)
        if circuit_breaker_ttl:
            blacklisted_until = _circuit_breaker.get(key)
            if blacklisted_until is not None and now < blacklisted_until:
                continue
        try:
            sock = _socket.create_connection((proxy.host, port), timeout=timeout)
        except OSError:
            if circuit_breaker_ttl:
                _circuit_breaker[key] = now + circuit_breaker_ttl
            continue
        sock.close()
        return proxy
    raise LookupError("no reachable proxy in the given list")

get_default_port(scheme)

Return the conventional port for a scheme, or None if unknown.

Source code in src/proxylib/netutils.py

def get_default_port(scheme: str) -> "int|None":
    """Return the conventional port for a scheme, or None if unknown."""
    scheme = scheme.lower()
    if scheme in _DEFAULT_PORTS:
        return _DEFAULT_PORTS[scheme]
    try:
        return _socket.getservbyname(scheme)
    except OSError:
        return None

get_ip(address)

Resolve a hostname or literal address to an ip_address, or None.

Source code in src/proxylib/netutils.py

def get_ip(address: str) -> "_ip.IPv4Address|_ip.IPv6Address|None":
    """Resolve a hostname or literal address to an ip_address, or None."""
    try:
        try:
            return _ip.ip_address(address)
        except ValueError:
            return _ip.ip_address(_socket.gethostbyname(address))
    except (ValueError, OSError):
        return None

get_local_interfaces(cache_ttl=10.0)

Return this host's local network interfaces, cached for cache_ttl seconds.

Pass cache_ttl=0/None to force a fresh enumeration.

Source code in src/proxylib/netutils.py

def get_local_interfaces(cache_ttl: "float|None" = 10.0) -> "List[_Interface]":
    """Return this host's local network interfaces, cached for ``cache_ttl`` seconds.

    Pass ``cache_ttl=0``/``None`` to force a fresh enumeration.
    """
    global _interfaces_cache
    if cache_ttl:
        cached = _interfaces_cache
        if cached is not None and (_time.monotonic() - cached[0]) < cache_ttl:
            return cached[1]
    result = _enumerate_interfaces()
    if cache_ttl:
        _interfaces_cache = (_time.monotonic(), result)
    return result

is_loopback_or_link_local(ip)

True for loopback (127/8, ::1) or link-local (169.254/16, fe80::/10) addresses -- these can never usefully go through a proxy (link-local is inherently confined to the local link). Shared by env.py's <local> NO_PROXY handling and ConfigurableProxyMap's bypass_local= option so the definition only lives in one place.

Source code in src/proxylib/netutils.py

def is_loopback_or_link_local(ip: "_ip.IPv4Address|_ip.IPv6Address") -> bool:
    """True for loopback (``127/8``, ``::1``) or link-local (``169.254/16``,
    ``fe80::/10``) addresses -- these can never usefully go through a proxy
    (link-local is inherently confined to the local link). Shared by
    ``env.py``'s ``<local>`` `NO_PROXY` handling and
    ``ConfigurableProxyMap``'s ``bypass_local=`` option so the definition
    only lives in one place.
    """
    return ip.is_loopback or ip.is_link_local