Python Async 装饰器保留输入

Question

对于以下文件：

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 错误。

这是在 python3.7 上，mypy 0.790

Answer 1

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