用ContinuationError取消函数执行

Question

虽然控制流抛馀量是一个有争议的话题，但是有一些非常流行的使用这种反模式的例子，比如C#'s异步例程，抛出OperationCanceledException来取消任务，或者用python抛出StopIteration来控制迭代器。

我想我将尝试在我的GitHub(记录器装潢包)中使用这样的异常。我叫它ContinuationError。通常，装饰器处理以下状态的日志记录

• 输入函数前的started
• 当函数成功执行时，completed
• 函数过早退出时的canceled
• 当意外发生时，faulted

异常通过将垃圾记录替换为错误来支持canceled状态：

在此之前：

logger.canceled(reason="No luck!")
return 5

之后：

raise ContinuationError("No luck!", 5)

它希望有一个理由来解释为什么取消是必要的，如果函数被期望返回某些内容，也可以选择返回值，还可以选择其他参数，这些参数随后被呈现为json消息。

class ContinuationError(Exception):
    """Raise this error to gracefully handle a function cancellation."""
    def __new__(cls, *args, **details) -> Any:
        instance = super().__new__(cls)
        instance.details = details | dict(reason=args[0])
        if len(args) > 1:
            instance.result = args[1]
        return instance

    def __init__(self, message: str, result: Optional[Any] = None, **details):
        super().__init__(message)

装饰师通过检查是否提供了result来处理它，并在必要时返回它。装饰器还提供了用于创建started详细信息的lambda或记录结果的另一个lambda。

def telemetry(on_started: Optional[OnStarted] = None, on_completed: Optional[OnCompleted] = None, **kwargs):
    """Provides flow telemetry for the decorated function."""

    on_started = on_started or (lambda _: {})
    on_completed = on_completed or (lambda _: {})

    def factory(decoratee):
        @contextlib.contextmanager
        def logger_scope() -> Logger:
            logger = Logger(
                module=inspect.getmodule(decoratee).__name__,
                scope=decoratee.__name__,
                attachment=kwargs.pop("attachment", None),
                parent=_scope.get()
            )

            token = _scope.set(logger)
            try:
                yield logger
            except Exception:
                logger.faulted()
                raise
            finally:
                _scope.reset(token)

        def inject_logger(logger: Logger, d: Dict):
            """ Injects Logger if required. """
            for n, t in inspect.getfullargspec(decoratee).annotations.items():
                if t is Logger:
                    d[n] = logger

        def params(*decoratee_args, **decoratee_kwargs) -> Dict[str, Any]:
            # Zip arg names and their indexes up to the number of args of the decoratee_args.
            arg_pairs = zip(inspect.getfullargspec(decoratee).args, range(len(decoratee_args)))
            # Turn arg_pairs into a dictionary and combine it with decoratee_kwargs.
            return {t[0]: decoratee_args[t[1]] for t in arg_pairs} | decoratee_kwargs

        if asyncio.iscoroutinefunction(decoratee):
            @functools.wraps(decoratee)
            async def decorator(*decoratee_args, **decoratee_kwargs):
                with logger_scope() as scope:
                    inject_logger(scope, decoratee_kwargs)
                    scope.started(**on_started(params(*decoratee_args, **decoratee_kwargs)))
                    try:
                        result = await decoratee(*decoratee_args, **decoratee_kwargs)
                        scope.completed(**on_completed(result))
                        return result
                    except ContinuationError as e:
                        if hasattr(e, "result"):
                            scope.canceled(**(on_completed(e.result) | e.details))
                            return e.result
                        else:
                            scope.canceled(**e.details)
        else:
            @functools.wraps(decoratee)
            def decorator(*decoratee_args, **decoratee_kwargs):
                with logger_scope() as scope:
                    inject_logger(scope, decoratee_kwargs)
                    scope.started(**on_started(params(*decoratee_args, **decoratee_kwargs)))
                    try:
                        result = decoratee(*decoratee_args, **decoratee_kwargs)
                        scope.completed(**on_completed(result))
                        return result
                    except ContinuationError as e:
                        if hasattr(e, "result"):
                            scope.canceled(**(on_completed(e.result) | e.details))
                            return e.result
                        else:
                            scope.canceled(**e.details)

        decorator.__signature__ = inspect.signature(decoratee)
        return decorator

    return factory

稍后，其中一个日志API检查异常并决定是记录正常消息还是记录实际错误：

    def _log(self, **kwargs):
        status = inspect.stack()[1][3]
        details = Logger.serialize_details(**kwargs)
        with _create_log_record(
                functools.partial(_set_module_name, name=self.module),
                functools.partial(_set_func_name, name=self.scope)
        ):
            # Ignore the ContinuationError as an actual error.
            is_error = all(sys.exc_info()) and sys.exc_info()[0] is not ContinuationError
            self._logger.log(level=self._logger.level, msg=None, exc_info=is_error, extra={
                "parent": self.parent.id if self.parent else None,
                "node": self.id,
                "status": status,
                "elapsed": self.elapsed,
                "details": details,
                "attachment": self.attachment
            })

在内部，包正在使用python的标准日志库。

示例

我就是这样用的：

import wiretap.src.wiretap as wiretap # becasue it's from the test environment

@wiretap.telemetry(on_started=lambda p: {"value": p["value"], "bar": p["bar"]}, on_completed=lambda r: {"count": r})
def foo(value: int, logger: wiretap.Logger = None, **kwargs) -> int:
    logger.running(test=f"{value}")
    raise wiretap.ContinuationError("No luck!", 0, foo="bar")
    return 3


if __name__ == "__main__":
    print(foo(1, bar="baz")) # <-- prints: 0

你觉得这个主意怎么样？我想我可能应该检查修饰函数是否会返回某些内容，并在未提供返回值时抛出无效的操作异常。

Answer 1

你的装潢师几乎没有字体提示。您可以使用ParamSpec提示您的代码。

下面是一个类型化的闭包包装装饰器：

import functools
from typing import Callable, ParamSpec, TypeVar

P = ParamSpec("P")
TRet = TypeVar("TRet")


def typed_decorator() -> Callable[[Callable[P, TRet]], Callable[P, TRet]]:
    def wrapper(fn: Callable[P, TRet]) -> Callable[P, TRet]:
        @functools.wraps(fn)
        def inner(*args: P.args, **kwargs: P.kwargs) -> TRet:
            return fn(*args, **kwargs)
        return inner
    return wrapper


def untyped_decorator():
    def wrapper(fn):
        @functools.wraps(fn)
        def inner(*args, **kwargs):
            return fn(*args, **kwargs)
        return inner
    return wrapper


@typed_decorator()
def foo(bar: str) -> int:
    return 1


@untyped_decorator()
def bar(bar: str) -> int:
    return 1


reveal_type(foo)
reveal_type(bar)

$ mypy --strict foo.py
foo.py:36: note: Revealed type is "def (bar: builtins.str) -> builtins.int"
foo.py:37: note: Revealed type is "Any"

正如您所看到的，我们基本上可以说参数是P，而返回类型只是TRet。

接下来，让DI正常工作。我将复制和修改您的inject_logger代码。因为我现在关注的是类型提示，所以我不得不要求您忽略函数的内容。函数做的事情和你的一样，DI记录器。

我将修改foo，将Logger作为参数，并使用默认设置。我们需要为类型提示按预期工作提供一个默认设置。否则，您将尝试在没有记录器的情况下调用该函数，并得到一个错误。

记录器有一些无稽之谈的代码，仅仅是为了确定我们与哪个记录器交互更容易推理。

import functools
import inspect
from typing import TYPE_CHECKING, Any, Callable, ParamSpec, TypeVar

P = ParamSpec("P")
TRet = TypeVar("TRet")
TArgs = TypeVar("TArgs", bound=tuple)  # type: ignore
TKwargs = TypeVar("TKwargs", bound=dict)  # type: ignore


class Logger:
    ID = 0
    id: int

    def __init__(self) -> None:
        self.id = self.ID
        type(self).ID += 1

    def __repr__(self) -> str:
        return f"<{type(self).__name__} {self.id}>"


def inject_logger(
    fn: Callable[..., Any],
    logger: Logger,
    args: TArgs,
    kwargs: TKwargs,
) -> tuple[TArgs, TKwargs]:
    """ Injects Logger if required."""
    sig = inspect.signature(fn)
    bound = sig.bind_partial(*args, **kwargs)
    for param in sig.parameters.values():
        if issubclass(Logger, param.annotation):
            bound.arguments.setdefault(param.name, logger)
    bound.apply_defaults()
    return bound.args, bound.kwargs  # type: ignore


def typed_decorator() -> Callable[[Callable[P, TRet]], Callable[P, TRet]]:
    def wrapper(fn: Callable[P, TRet]) -> Callable[P, TRet]:
        @functools.wraps(fn)
        def inner(*args: P.args, **kwargs: P.kwargs) -> TRet:
            args, kwargs = inject_logger(fn, Logger(), args, kwargs)
            print(args, kwargs)
            return fn(*args, **kwargs)

        inner.__signature__ = fn.__signature__ = inspect.signature(fn)  # type: ignore
        return inner
    return wrapper


@typed_decorator()
def foo(bar: str, logger: Logger = Logger(), /) -> int:
    return 1


if TYPE_CHECKING:
    reveal_type(foo)

if __name__ == "__main__":
    foo("test")

$ python foo.py
('test', ) {}
$ mypy --strict foo.py
foo.py:58: note: Revealed type is "def (builtins.str, foo.Logger =) -> builtins.int"

我们可以通过使用Concatenate来避免指定默认值的需要。Concatenate的限制是语法只适用于函数左侧的参数。

def typed_decorator() -> Callable[[Callable[Concatenate[Logger, P], TRet]], Callable[P, TRet]]:
    def wrapper(fn: Callable[Concatenate[Logger, P], TRet]) -> Callable[P, TRet]:
        @functools.wraps(fn)
        def inner(*args: P.args, **kwargs: P.kwargs) -> TRet:
            return fn(Logger(), *args, **kwargs)

        fn.__signature__ = sig = inspect.signature(fn)  # type: ignore
        params = iter(sig.parameters.values())
        next(params, None)
        inner.__signature__ = inspect.Signature(list(params), return_annotation=sig.return_annotation)  # type: ignore
        return inner
    return wrapper


@typed_decorator()
def foo(logger: Logger, bar: str) -> int:
    return 1


if TYPE_CHECKING:
    reveal_type(foo)

$ mypy --strict foo.py
foo.py:56: note: Revealed type is "def (bar: builtins.str) -> builtins.int"

我们可以通过使用带有Protocol的overload来支持这两个选项。正如您可能从增加的# type: ignore注释中看到的那样，这种方法在内部有些简陋。

def inject_logger(
    fn: Callable[..., Any],
    logger: Logger,
    args: TArgs,
    kwargs: TKwargs,
) -> tuple[TArgs, TKwargs]:
    """ Injects Logger if required."""
    sig = inspect.signature(fn)
    if ((param := next(iter(sig.parameters.values()), None)) is not None
        and issubclass(param.annotation, Logger)
    ):
        args = (logger,) + args  # type: ignore
    bound = sig.bind_partial(*args, **kwargs)
    for param in sig.parameters.values():
        if issubclass(param.annotation, Logger):
            bound.arguments.setdefault(param.name, logger)
    bound.apply_defaults()
    return bound.args, bound.kwargs  # type: ignore


class PTypedDecorator(Protocol):
    @overload
    def __call__(self, /, fn: Callable[Concatenate[Logger, P], TRet]) -> Callable[P, TRet]: ...
    @overload
    def __call__(self, /, fn: Callable[P, TRet]) -> Callable[P, TRet]: ...


def typed_decorator() -> PTypedDecorator:
    def wrapper(fn: Callable[Concatenate[Logger, P], TRet] | Callable[P, TRet]) -> Callable[P, TRet]:
        @functools.wraps(fn)
        def inner(*args: P.args, **kwargs: P.kwargs) -> TRet:
            args, kwargs = inject_logger(fn, Logger(), args, kwargs)
            print(args, kwargs)
            return fn(*args, **kwargs)

        fn.__signature__ = sig = inspect.signature(fn)  # type: ignore
        params = iter(sig.parameters.values())
        if ((param := next(params, None)) is not None
            and issubclass(param.annotation, Logger)
        ):
            sig = inspect.Signature(list(params), return_annotation=sig.return_annotation)
        inner.__signature__ = sig  # type: ignore
        return inner
    return wrapper


@typed_decorator()
def foo(logger: Logger, bar: str, /) -> int:
    return 1


@typed_decorator()
def bar(bar: str, *, logger: Logger = Logger()) -> int:
    return 1


if TYPE_CHECKING:
    reveal_type(foo)
    reveal_type(bar)

if __name__ == "__main__":
    print(inspect.signature(foo))
    foo("test")
    print(inspect.signature(bar))
    bar("test")

$ python foo.py
(bar: str, /) -> int
(, 'test') {}
(bar: str, *, logger: __main__.Logger = ) -> int
('test',) {'logger': }
$ mypy --strict foo.py
foo.py:73: note: Revealed type is "def (bar: builtins.str) -> builtins.int"
foo.py:74: note: Revealed type is "def (bar: builtins.str, *, logger: foo.Logger =) -> builtins.int"

评论

• 您的现有代码并不真正支持类型提示。
• @wiretap.telemetry(on_started=lambda p：{“on_started=lambda”：p，"bar"：p}，on_completed=lambda r：{"count"：r}) def (值: int，logger: wiretap.Logger = None，**kwargs) -> int：logger: wiretap.Logger = None要正确键入的唯一原因是如果您没有在严格模式下运行静态分析工具。默认情况下，mypy在非严格模式下运行。这种模式对于显示突出的类型问题很有用，比如将int传递给str的函数。好处是通过隐藏一些更迂腐的静态分析工具，拥有大量Python代码库的用户可以轻松地进入静态Python。
• 您的factory闭包中有很多函数可以是全局的，看看我的inject_logger。
• 当前的inject_logger不能只支持位置参数。但是，要添加记录器，您可以对kwargs进行变异，记录器可以在/之前提供，成为一个位置惟一的参数。
• on_started或on_completed都没有正确的类型。我可以理解，您可能认为在*中使用lambda和在D44中使用**对于on_started来说可能很难看。因此，具有非类型化的接口可能更好。>>> (lambda *_，foo，**__：foo)(foo="foo") 'foo‘>>> (lambda *_，foo，**__：foo)(bar="foo")回溯(最近一次调用)：文件""，第1行，在 TypeError：()缺少一个必需关键字-只有参数：'foo’

如果您忽略输入，您的代码是好的。在我看来，这份工作似乎设计得不太好。

正如您可能知道的那样，在Python中，需要使用异常来控制流。但是，对于在这里使用异常来控制流是否很好，我真的不能发表评论。我需要查看您的库的实际使用情况，看看另一个选项是否会更好。但是，您在这里使用异常似乎是获得您希望通过装饰器实现的一个好方法。另一种方法是返回一个与ContinuationError完全相同的接口的D49。