railtracks.prebuilt.middleware.post_verifier

  1from __future__ import annotations
  2
  3import asyncio
  4import functools
  5import inspect
  6from typing import Awaitable, Callable, Concatenate, ParamSpec, TypeVar, overload
  7
  8from railtracks.events.middleware import (
  9    MiddlewareVerifierPostFailureEvent,
 10    MiddlewareVerifierPostInvocationEvent,
 11    MiddlewareVerifierPostResponseEvent,
 12)
 13from railtracks.events.send import emit
 14from railtracks.middleware.core import Middleware, wrap_node
 15from railtracks.middleware.verdict import (
 16    Verdict,
 17    VerifierDecision,
 18    VerifierRejectedError,
 19)
 20from railtracks.utils.logging.create import get_rt_logger
 21from railtracks.utils.unpack import unpack_async_sync
 22
 23logger = get_rt_logger(__name__)
 24
 25_P = ParamSpec("_P")
 26_R = TypeVar("_R")
 27
 28_ApproveFn = Callable[Concatenate[_R, _P], Verdict[_R] | Awaitable[Verdict[_R]]]
 29
 30_POSITIONAL_KINDS = (
 31    inspect.Parameter.POSITIONAL_ONLY,
 32    inspect.Parameter.POSITIONAL_OR_KEYWORD,
 33)
 34
 35
 36def _require_result_first(approve_fn: Callable) -> None:
 37    params = list(inspect.signature(approve_fn).parameters.values())
 38    first_ok = (
 39        params and params[0].name == "result" and params[0].kind in _POSITIONAL_KINDS
 40    )
 41
 42    if not first_ok:
 43        got = params[0].name if params else "no parameters"
 44        message = (
 45            "post_verifier's approve_fn must take `result` as its first "
 46            f"positional parameter, got {got!r}. post_verifier calls "
 47            "approve_fn(result, *args, **kwargs), e.g.:\n"
 48            "    def approve(result, *args, **kwargs) -> Verdict: ..."
 49        )
 50        raise TypeError(message)
 51
 52
 53@overload
 54def post_verifier(
 55    approve_fn: _ApproveFn[_R, _P],
 56    /,
 57    *,
 58    timeout: float | None = None,
 59    name: str | None = None,
 60) -> Middleware[_P, _R]: ...
 61@overload
 62def post_verifier(
 63    *, timeout: float | None = None, name: str | None = None
 64) -> Callable[[_ApproveFn[_R, _P]], Middleware[_P, _R]]: ...
 65
 66
 67def post_verifier(
 68    approve_fn: _ApproveFn[_R, _P] | None = None,
 69    /,
 70    *,
 71    timeout: float | None = None,
 72    name: str | None = None,
 73) -> Middleware[_P, _R] | Callable[[_ApproveFn[_R, _P]], Middleware[_P, _R]]:
 74    """Build a node-verification middleware around ``approve_fn`` that gates a
 75    call's OUTPUT AFTER it has already run.
 76
 77    The wrapped node always runs first. ``approve_fn`` is then called with the
 78    produced value as its first positional parameter, followed by the node's
 79    own ``*args, **kwargs`` — sync or async, both supported — and must return
 80    a `Verdict`. This shape is validated eagerly, at ``post_verifier(...)``
 81    call time: an ``approve_fn`` that doesn't take ``result`` first raises
 82    `TypeError` immediately, naming what was found instead.
 83
 84    Decline can't undo the call (it already happened) but still raises
 85    `VerifierRejectedError`, stopping the result from propagating onward. On
 86    accept, the result propagates using the verdict's ``result`` if it
 87    supplied an override, otherwise the original result unchanged.
 88
 89    If ``timeout`` is set and ``approve_fn`` doesn't respond in time, the call
 90    is treated as declined with reason ``"timeout"``.
 91
 92    See also :func:`~railtracks.prebuilt.middleware.pre_verifier.pre_verifier`,
 93    which gates whether a call happens at all, BEFORE it runs. For the full
 94    picture (composing with other middleware, custom approval backends,
 95    guided walkthroughs), see the Verifiers docs.
 96    """
 97
 98    if approve_fn is None:
 99        return lambda fn: post_verifier(fn, timeout=timeout, name=name)
100
101    _require_result_first(approve_fn)
102    return wrap_node(_wrapper(approve_fn, timeout), name=name)
103
104
105def _wrapper(approve_fn: _ApproveFn[_R, _P], timeout: float | None):
106    @functools.wraps(approve_fn)
107    async def wrapped(
108        call: Callable[_P, Awaitable[_R]], *args: _P.args, **kwargs: _P.kwargs
109    ) -> _R:
110        result = await call(*args, **kwargs)
111
112        await emit(MiddlewareVerifierPostInvocationEvent(response=result))
113
114        is_timeout = False
115        try:
116            review = unpack_async_sync(approve_fn(result, *args, **kwargs))
117            if timeout is None:
118                verdict = await review
119            else:
120                verdict = await asyncio.wait_for(review, timeout=timeout)
121        except asyncio.TimeoutError:
122            verdict = Verdict(accepted=False, comment="timeout")
123            is_timeout = True
124        except Exception as e:
125            await emit(MiddlewareVerifierPostFailureEvent.from_exception(e))
126            raise
127
128        new_result = verdict.result if verdict.result is not None else result
129        overridden = verdict.result is not None
130
131        await emit(
132            MiddlewareVerifierPostResponseEvent(
133                decision=VerifierDecision.from_verdict(
134                    verdict, overridden=overridden, timeout=is_timeout
135                ),
136                response=new_result,
137            )
138        )
139
140        if not verdict.accepted:
141            raise VerifierRejectedError(verdict.comment or "rejected")
142
143        if verdict.comment:
144            logger.info("post_verifier accepted with comment: %s", verdict.comment)
145
146        return new_result
147
148    return wrapped