syshttp.hooks

syshttp.hooks

URL hooks: a wrapper a curl front end runs, or a callable syshttp calls, for requests whose URL matches a pattern -- declared in the environment, each variable read under the prefixes of an :class:~syshttp.env.HTTPEnv. SYSHTTP_HOOKS_<KEY> (plural) declares a hook; SYSHTTP_HOOK_<NAME> (singular) is a variable one hook RUN exports to its wrapper.

SYSHTTP_HOOKS_<KEY>=<regex>               matched (re.search) against the URL
                                          the caller asked for -- the origin URL,
                                          never the proxy's or a gateway's rewrite
SYSHTTP_HOOKS_<KEY>_CURL=<command>        a curl front end runs THIS instead of
                                          itself, with its own argv appended
SYSHTTP_HOOKS_<KEY>_PY=<module:callable>  syshttp calls it before the request:
                                          callable(client, method, url, kwargs)

A hook does what it wants with the client (log in, set a header in kwargs["headers"], refresh a token in the jar) and returns None to let the request proceed; anything else it returns is used AS the response (an HTTPResponse, cached or synthetic). kwargs is the keyword arguments the HTTPClient.request() call gave (data, headers, stream, allow_redirects, timeout, deadline, auth), mutable in place. <module:callable> is importable from sys.path or a <path>.py:callable file.

The curl wrapper is a shell-quoted command line -- a program and its own arguments -- that the front end's whole argv is appended to::

SYSHTTP_HOOKS_X_CURL='cmd fetch --'   ->   cmd fetch -- "$@"

Quote (single or double) what has spaces; a backslash is literal on every platform, so a Windows path needs no doubling. A first word that is a .py file runs under the front end's interpreter. The wrapper receives (:func:wrapper_environment) SYSHTTP_HOOK_URL (the requested URL, so it need not parse argv), SYSHTTP_HOOK_KEY (which hook matched), SYSHTTP_HOOK_CURL naming the front end so it can call curl back after its own work, and SYSHTTP_HOOK_SKIP carrying its KEY, so that call does not run the same wrapper again. A hook with only a _PY target is ignored by a curl front end, one with only _CURL by syshttp.

Several hooks may match one URL: syshttp calls each in KEY order (sorted), the first returning a response ends the chain; a curl front end runs the first _CURL in that order. A pattern that does not compile is a configuration error (ValueError naming the variable) the first time it would be consulted.

Hook(key, pattern, curl, py, variable=None)

Bases: object

One declared hook. variable is the name of the variable declaring it (SYSHTTP_HOOKS_X), which every message about the hook names.

Source code in src/syshttp/hooks.py

def __init__(
    self,
    key: str,
    pattern: "re.Pattern[str]",
    curl: Optional[str],
    py: Optional[str],
    variable: Optional[str] = None,
):
    self.key = key
    self.pattern = pattern
    self.curl = curl
    self.py = py
    self.variable = variable or HOOKS + key

call_py_hooks(client, method, url, kwargs, *, env=None)

Run every _PY hook matching url in order; the first non-None result is returned (the caller uses it as the response), else None.

Source code in src/syshttp/hooks.py

def call_py_hooks(
    client: "HTTPClient",
    method: str,
    url: str,
    kwargs: Dict[str, Any],
    *,
    env: Optional[HTTPEnv] = None,
) -> "Optional[HTTPResponse]":
    """Run every ``_PY`` hook matching ``url`` in order; the first non-None
    result is returned (the caller uses it as the response), else ``None``."""
    for hook in matching(url, kind="py", env=env):
        if not hook.py:  # `kind="py"` already filtered; for the type checker
            continue
        try:
            fn = load_callable(hook.py)
            result = fn(client, method, url, kwargs)
        except Exception as exc:
            # The variable, so the failure is traceable to its declaration;
            # the URL without its userinfo, query and fragment, which carry
            # secrets. The hook's own exception is chained as it is.
            raise HookError(
                "%s%s (%s) failed for %s %s: %s" % (hook.variable, PY_SUFFIX, hook.py, method, _bare_url(url), exc)
            ) from exc
        if result is not None:
            return result
    return None

curl_command(hook, python=None)

The argv prefix a _CURL target means, the front end's argv to be appended: the shell-quoted command line split; a bare path to an existing file (spaces and all) is that one program; a first word that is a .py file runs under python (default: this interpreter).

Source code in src/syshttp/hooks.py

def curl_command(hook: Hook, python: Optional[str] = None) -> List[str]:
    """The argv prefix a ``_CURL`` target means, the front end's argv to be
    appended: the shell-quoted command line split; a bare path to an existing
    file (spaces and all) is that one program; a first word that is a ``.py``
    file runs under ``python`` (default: this interpreter)."""
    value = (hook.curl or "").strip()
    if not value:
        raise ConfigurationError("%s%s is empty" % (hook.variable, CURL_SUFFIX))
    argv = [value] if Path(value).is_file() else split_command(value)
    if not argv:
        raise ConfigurationError("%s%s holds no command" % (hook.variable, CURL_SUFFIX))
    if argv[0].lower().endswith(".py") and Path(argv[0]).is_file():
        argv = [python or sys.executable, *argv]
    return argv

hooks_from_env(*, env=None)

Every configured hook, in KEY order. KEYs are upper-cased (variable names are compared as Windows compares them), so SYSHTTP_HOOKS_mykey and SYSHTTP_HOOKS_MYKEY are one hook; the first prefix that declares a variable of the hook wins for that variable. A _CURL/_PY variable whose KEY has no pattern is ignored (nothing to match), and so is a blank target.

Source code in src/syshttp/hooks.py

def hooks_from_env(*, env: Optional[HTTPEnv] = None) -> List[Hook]:
    """Every configured hook, in KEY order. KEYs are upper-cased (variable
    names are compared as Windows compares them), so ``SYSHTTP_HOOKS_mykey``
    and ``SYSHTTP_HOOKS_MYKEY`` are one hook; the first prefix that declares a
    variable of the hook wins for that variable. A ``_CURL``/``_PY`` variable whose KEY has
    no pattern is ignored (nothing to match), and so is a blank target."""
    env = _env(env)
    found = {rest.upper(): (value, variable) for rest, value, variable in env.family(HOOKS)}
    hooks = []
    for key in sorted(found):
        if key.endswith((CURL_SUFFIX, PY_SUFFIX)):
            continue
        value, variable = found[key]
        try:
            pattern = re.compile(value)
        except re.error as exc:
            raise ConfigurationError("%s is not a valid regular expression: %s" % (variable, exc))
        curl = found.get(key + CURL_SUFFIX, (None, None))[0]
        py = found.get(key + PY_SUFFIX, (None, None))[0]
        if curl or py:
            hooks.append(Hook(key, pattern, curl, py, variable))
    return hooks

load_callable(spec)

module:attr (importable) or <file>.py:attr, which must be callable (:func:load_object).

Source code in src/syshttp/hooks.py

def load_callable(spec: str) -> "PyHook":
    """``module:attr`` (importable) or ``<file>.py:attr``, which must be
    callable (:func:`load_object`)."""
    obj = load_object(spec, "a hook callable")
    if not callable(obj):
        raise ConfigurationError("%s is not callable" % spec)
    return obj

load_object(spec, what='an object')

The object module:attr (importable) or <file>.py:attr names; attr may be dotted. A file is loaded once per process, by its resolved path. A malformed spec is a ConfigurationError naming what; a module or attribute that is not there raises as the import does.

Source code in src/syshttp/hooks.py

def load_object(spec: str, what: str = "an object") -> Any:
    """The object ``module:attr`` (importable) or ``<file>.py:attr`` names;
    ``attr`` may be dotted. A file is loaded once per process, by its resolved
    path. A malformed ``spec`` is a ``ConfigurationError`` naming ``what``; a
    module or attribute that is not there raises as the import does."""
    target, sep, attr = spec.rpartition(":")
    if not sep or not target or not attr:
        raise ConfigurationError("%s is written <module or file.py>:<name>, not %r" % (what, spec))
    if target.endswith(".py") and Path(target).is_file():
        module = _file_module(Path(target))
    else:
        module = importlib.import_module(target)
    obj: Any = module
    for part in attr.split("."):
        obj = getattr(obj, part)
    return obj

matching(url, *, kind=None, env=None)

The hooks for url (the caller's URL), KEY order, minus the skipped ones; kind ("curl" / "py") keeps only hooks with that target.

Source code in src/syshttp/hooks.py

def matching(url: str, *, kind: Optional[str] = None, env: Optional[HTTPEnv] = None) -> List[Hook]:
    """The hooks for ``url`` (the caller's URL), KEY order, minus the skipped
    ones; ``kind`` (``"curl"`` / ``"py"``) keeps only hooks with that target."""
    skip = skipped(env=env)
    out = []
    for hook in hooks_from_env(env=env):
        if hook.key.upper() in skip or not hook.matches(url):
            continue
        if kind and not getattr(hook, kind):
            continue
        out.append(hook)
    return out

skipped(*, env=None)

The KEYs SYSHTTP_HOOK_SKIP (a comma list) says not to run, upper-cased.

Source code in src/syshttp/hooks.py

def skipped(*, env: Optional[HTTPEnv] = None) -> "set[str]":
    """The KEYs ``SYSHTTP_HOOK_SKIP`` (a comma list) says not to run, upper-cased."""
    return {k.strip().upper() for k in (_env(env).get(HOOK_SKIP) or "").split(",") if k.strip()}

split_command(value)

A shell-quoted command line into argv: single and double quotes group words with spaces, a backslash is literal (no escaping on any platform, so C:\Tools\wrap.exe stays what it is), # starts no comment.

Source code in src/syshttp/hooks.py

def split_command(value: str) -> List[str]:
    """A shell-quoted command line into argv: single and double quotes group
    words with spaces, a backslash is literal (no escaping on any platform,
    so ``C:\\Tools\\wrap.exe`` stays what it is), ``#`` starts no comment."""
    lex = shlex.shlex(value, posix=True)
    lex.whitespace_split = True
    lex.escape = ""
    lex.commenters = ""
    return list(lex)

wrapper_environment(hook, url, entry, *, env=None)

The variables a _CURL wrapper run adds to its environment: SYSHTTP_HOOK_URL (url), SYSHTTP_HOOK_KEY, SYSHTTP_HOOK_CURL (entry, the front end to call back) and SYSHTTP_HOOK_SKIP with this KEY added, so the call back does not run the wrapper again; each under the first prefix.

Source code in src/syshttp/hooks.py

def wrapper_environment(hook: Hook, url: str, entry: str, *, env: Optional[HTTPEnv] = None) -> Dict[str, str]:
    """The variables a ``_CURL`` wrapper run adds to its environment:
    ``SYSHTTP_HOOK_URL`` (``url``), ``SYSHTTP_HOOK_KEY``, ``SYSHTTP_HOOK_CURL``
    (``entry``, the front end to call back) and ``SYSHTTP_HOOK_SKIP`` with this
    KEY added, so the call back does not run the wrapper again; each under the
    first prefix."""
    env = _env(env)
    url_var, key_var, curl_var, skip_var = (env.name(s) for s in (HOOK_URL, HOOK_KEY, HOOK_CURL, HOOK_SKIP))
    skip = ",".join(sorted(skipped(env=env) | {hook.key.upper()}))
    return {url_var: url, key_var: hook.key, curl_var: entry, skip_var: skip}