时间戳

首页 > 解决方案 > Python Async 装饰器保留输入

python - Python Async 装饰器保留输入

问题描述

对于以下文件：

from abc import ABC, abstractmethod
from typing import Any, Awaitable, Callable, TypeVar, cast

T = TypeVar('T')


def dec(args: Any) -> Callable[..., Awaitable[T]]:
    def dec2(f: Callable[..., Awaitable[T]]) -> Awaitable[T]:
        async def wrapper(*args:Any, **kwargs:Any) -> T:
            print(args)

            return cast(T, await f(*args, **kwargs))
        return cast(Awaitable[T], wrapper)
    return dec2


class A(ABC):
    @abstractmethod
    async def fn(self) -> 'bool':
        pass

class B(A):
    @dec('hi')
    async def fn(self) -> 'bool':
        return True


class C(A):
    @dec('hi')
    async def fn(self) -> 'bool':
        return False

我收到以下 mypy 错误：

$ mypy typetest.py
typetest.py:24: error: Signature of "fn" incompatible with supertype "A"
typetest.py:30: error: Signature of "fn" incompatible with supertype "A"
Found 2 errors in 1 file (checked 1 source file)

键入需要如何工作才能保留类签名并且不会收到 mypy 错误。

这是在带有 mypy 0.790 的 python3.7 上

标签： pythonpython-3.xmypy

解决方案

TLDR：函数声明async def name(parameters) -> R:创建类型的对象(parameters) -> Awaitable[R]，而不是Awaitable[R]。这意味着强制转换cast(Awaitable[T], wrapper)是错误的并且应该被省略，并且Callable还必须调整各种返回类型。

显示async def函数执行的天真的装饰器（大致dec2）如下所示：

def show_call(f: Callable[..., Awaitable[T]]) -> Callable[..., Awaitable[T]]:
    async def wrapper(*args: Any, **kwargs: Any) -> T:
        print(f"Executing {f.__name__}")
        return await f(*args, **kwargs)
    return wrapper

这接收一个可调用的匹配async def并返回一个可调用的匹配async def。换句话说，它保留了“async函数”的一般类型。
请注意，cast不需要。

由于Callable参数被指定为...，因此参数信息丢失。这可以通过类似于类型变量的PEP 612 (Python 3.10 / ) 参数变量来解决。typing_extensions

from typing import Any, Awaitable, Callable, TypeVar, ParamSpec

T = TypeVar('T')    # the callable/awaitable return type
P = ParamSpec('P')  # the callable parameters


def dec(message: Any) -> Callable[[Callable[P, Awaitable[T]]], Callable[P, Awaitable[T]]]:
    def dec2(f: Callable[P, Awaitable[T]]) -> Callable[P, Awaitable[T]]:
        async def wrapper(*args: Any, **kwargs: Any) -> T:
            print(message)
            return await f(*args, **kwargs)
        return wrapper
    return dec2

推荐阅读