Patching registry

proxylib.patching

Extensible patching registry: globally wire an active ProxyMap into third-party HTTP clients (requests, urllib, ...) without every caller needing to know each integration's specific mounting API.

Built-in targets: "requests" (wraps requests.Session.__init__ so newly created sessions mount a ProxyMapAdapter for http:/// https:// -- sessions created before :func:patch are untouched) and "urllib" (installs a global opener via urllib.request.install_opener). Register your own with :func:register_patcher.

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

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

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

patch(proxymap, targets=None)

Globally wire proxymap into targets (default: every registered target).

Calling this while already patched replaces the previous patch (with a warning) rather than stacking -- there is one active resolver at a time, not a layered set.

Source code in src/proxylib/patching.py

def patch(proxymap: ProxyMap, targets: "Optional[List[str]]" = None) -> None:
    """Globally wire ``proxymap`` into ``targets`` (default: every registered target).

    Calling this while already patched **replaces** the previous patch
    (with a warning) rather than stacking -- there is one active resolver
    at a time, not a layered set.
    """
    global _active_map, _active_targets
    with _lock:
        if _active_map is not None:
            warnings.warn(
                "proxylib.patch() called while already patched -- replacing "
                "the previous patch instead of stacking.",
                stacklevel=2,
            )
            _unpatch_locked()

        names = list(targets) if targets is not None else list(_registry)
        applied: "List[str]" = []
        try:
            for name in names:
                if name not in _registry:
                    raise KeyError(f"No patcher registered for {name!r}")
                patch_func, _ = _registry[name]
                patch_func(proxymap)
                applied.append(name)
        except BaseException:
            # Partial failure: undo whatever *did* apply before propagating,
            # so a failed patch() doesn't leave a half-patched process.
            for name in reversed(applied):
                _registry[name][1]()
            raise

        _active_map = proxymap
        _active_targets = applied

register_patcher(name, patch_func, unpatch_func)

Register a named integration for :func:patch/:func:unpatch to drive.

patch_func(proxymap) applies the integration; unpatch_func() undoes it (no arguments -- capture whatever state you need in a closure). Re-registering an existing name replaces it.

Source code in src/proxylib/patching.py

def register_patcher(name: str, patch_func: PatchFunc, unpatch_func: UnpatchFunc) -> None:
    """Register a named integration for :func:`patch`/:func:`unpatch` to drive.

    ``patch_func(proxymap)`` applies the integration; ``unpatch_func()``
    undoes it (no arguments -- capture whatever state you need in a
    closure). Re-registering an existing ``name`` replaces it.
    """
    _registry[name] = (patch_func, unpatch_func)

unpatch()

Undo whatever :func:patch last applied, in reverse order.

Idempotent: calling it when nothing is patched is a no-op.

Source code in src/proxylib/patching.py

def unpatch() -> None:
    """Undo whatever :func:`patch` last applied, in reverse order.

    Idempotent: calling it when nothing is patched is a no-op.
    """
    with _lock:
        _unpatch_locked()