Python 异步装饰器保留输入

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