Patching registry

proxylib.patching

Wire proxy maps into HTTP clients: requests, urllib, httpx, aiohttp and pip built in, others through :func:register_patcher.

Each client gets proxylib's integration once, when it is imported, in any program of the environment: proxylib's importpatch entry points run at interpreter start, and importing proxylib registers the same hooks. The integration asks the client's active map: the newest :func:patch that covers it, else its default.

PROXYLIB_IMPORTPATCH chooses the clients that get the integration: unset, every built-in one (and each registered with unnamed=True); 1/true/yes/on, every registered one; or target names separated by : (requests:httpx; a client's module name, such as urllib.request, works too); 0/false/no/off, none until :func:patch is called. Its default map has no opinion, so the client chooses its proxy as it always does, and proxylib only reads what the client cannot: a PAC result (PROXY a:3128; PROXY b:3128; DIRECT) or a URL-prefix gateway given where the client takes a proxy.

PROXYLIB_AUTOPROXY (the same forms; unset or 0, none) chooses the clients whose default is :func:~proxylib.auto_proxy: the proxy variables, the OS resolver and settings, PAC and WPAD. A client named there is installed.

InstallFunc = Callable[[ProxyMap], None] module-attribute

UninstallFunc = Callable[[], None] module-attribute

__all__ = ('Patch', 'active', 'patch', 'register_patcher', 'targets', 'unpatch') module-attribute

Patch(proxymap, targets)

A :func:patch in effect. :meth:undo, or leaving its with block, makes the map active before it active again.

Source code in src/proxylib/patching.py

def __init__(self, proxymap: ProxyMap, targets: "Optional[FrozenSet[str]]") -> None:
    self.proxymap = proxymap
    self.targets = targets

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

active property

proxymap = proxymap instance-attribute

targets = targets instance-attribute

__enter__()

Source code in src/proxylib/patching.py

def __enter__(self) -> "Patch":
    return self

__exit__(*exc_info)

Source code in src/proxylib/patching.py

def __exit__(self, *exc_info: object) -> None:
    self.undo()

__repr__()

Source code in src/proxylib/patching.py

def __repr__(self) -> str:
    targets = "all" if self.targets is None else sorted(self.targets)
    return f"<Patch {self.proxymap!r} for {targets}>"

undo()

End this patch. Patches end newest first: undoing one that a later patch still covers raises RuntimeError. Undoing an ended patch does nothing.

Source code in src/proxylib/patching.py

def undo(self) -> None:
    """End this patch. Patches end newest first: undoing one that a later
    patch still covers raises ``RuntimeError``. Undoing an ended patch
    does nothing."""
    with _lock:
        if self not in _frames:
            return
        if _frames[-1] is not self:
            raise RuntimeError("patches are undone newest first; a later patch is still in effect")
        _frames.pop()

active(target)

The map target's client uses now.

Source code in src/proxylib/patching.py

def active(target: str) -> ProxyMap:
    """The map ``target``'s client uses now."""
    with _lock:
        for handle in reversed(_frames):
            if handle.targets is None or target in handle.targets:
                return handle.proxymap
    return _default_map(target)

patch(proxymap, *, targets=None)

Make proxymap the map of every client in targets (default: all registered), until the returned :class:Patch is undone or :func:unpatch is called. Patches nest: the newest wins.

A named target whose client is not installed raises :class:MissingExtraError; with the default, such clients are skipped. An unknown name raises ValueError.

Source code in src/proxylib/patching.py

def patch(proxymap: ProxyMap, *, targets: "Optional[Iterable[str]]" = None) -> Patch:
    """Make ``proxymap`` the map of every client in ``targets`` (default: all
    registered), until the returned :class:`Patch` is undone or
    :func:`unpatch` is called. Patches nest: the newest wins.

    A named target whose client is not installed raises
    :class:`MissingExtraError`; with the default, such clients are skipped.
    An unknown name raises ``ValueError``.
    """
    names = None if targets is None else frozenset(targets)
    for name in names or ():
        if name not in _registry:
            raise ValueError(f"no patch target {name!r}; registered: {tuple(_registry)}")
        top = _registry[name].module.partition(".")[0]
        if sys.modules.get(top) is None and importlib.util.find_spec(top) is None:
            raise MissingExtraError(f"the {name} patch target", name=top)
    handle = Patch(proxymap, names)
    with _lock:
        _frames.append(handle)
    for name in names if names is not None else list(_registry):
        if names is not None:
            importlib.import_module(_registry[name].module)
        _hook(name)
    # Clients already imported, which the hook saw before this patch wanted them.
    _install_ready()
    return handle

register_patcher(name, install, *, uninstall, module, unnamed=False, replace=False)

Register a client integration. install(proxymap) is called once, when module (the client) has been imported and the target is wanted (named in PROXYLIB_IMPORTPATCH or PROXYLIB_AUTOPROXY, patched from code, or -- with PROXYLIB_IMPORTPATCH unset -- unnamed=True: an integration that keeps the client's own behaviour while its map has no opinion), with a map that always answers as the active map does; uninstall() removes it. An existing name raises ValueError unless replace=True.

Source code in src/proxylib/patching.py

def register_patcher(
    name: str,
    install: InstallFunc,
    *,
    uninstall: UninstallFunc,
    module: str,
    unnamed: bool = False,
    replace: bool = False,
) -> None:
    """Register a client integration. ``install(proxymap)`` is called once,
    when ``module`` (the client) has been imported and the target is wanted
    (named in ``PROXYLIB_IMPORTPATCH`` or ``PROXYLIB_AUTOPROXY``, patched from
    code, or -- with ``PROXYLIB_IMPORTPATCH`` unset -- ``unnamed=True``: an
    integration that keeps the client's own behaviour while its map has no
    opinion), with a map that always answers as the active map does;
    ``uninstall()`` removes it. An existing ``name`` raises ``ValueError``
    unless ``replace=True``."""
    with _lock:
        if name in _registry and not replace:
            raise ValueError(f"patch target {name!r} is already registered")
        _registry[name] = _Target(module, install, uninstall, unnamed)
        _hooked.discard(name)
    _hook(name)

targets()

The registered patch targets.

Source code in src/proxylib/patching.py

def targets() -> "Tuple[str, ...]":
    """The registered patch targets."""
    with _lock:
        return tuple(_registry)

unpatch()

End every patch: each client is back on the default map.

Source code in src/proxylib/patching.py

def unpatch() -> None:
    """End every patch: each client is back on the default map."""
    with _lock:
        _frames.clear()