Python 修饰异步函数的注释

Question

我在为防止 aiohttp 错误而修饰的协程上使用注释时遇到困难。我有两个功能：

from typing import Callable, Awaitable, Optional
from os import sep
import aiofiles
import aiohttp
from asyncio.exceptions import TimeoutError
from aiohttp.client_exceptions import ClientError


def catch_aiohttp_errors(func: Callable[..., Awaitable]) -> Callable[..., Awaitable]:
    async def wrapper(*args):
        try:
            return await func(*args)
        except (TimeoutError, ClientError):
            return None
    return wrapper


@catch_aiohttp_errors
async def download(url: str, download_path: str, filename: str, suffix: str) -> Optional[str]:
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            async with aiofiles.open(download_path + sep + filename + '.' + suffix, 'wb') as file:
                async for chunk in response.content.iter_chunked(1024):
                    await file.write(chunk) if chunk else await file.write(b'')
    return download_path + sep + filename + '.' + suffix

实现装饰器功能的主要原因是我有几个使用 aiohttp 的异步函数，我不想在每个类似的函数中都写 try/except 语句。

我遇到的问题是我的第二个函数的正确注释。
如您所见，它 returns str。但是如果会报错，会根据装饰器函数的try/except部分returnNone。用 Optional[str] 注释这样的函数是否正确？

Answer 1

我建议使用 TypeVar 作为 Awaitable 类型参数以停止丢失有关装饰函数的信息：在您的示例中调用 download 的结果将是类型 Any.此外，使用 ParamSpec 将有助于保留参数。最后，这样的事情应该可行（假设 python 3.10，否则将所有未知的 typing 导入替换为 typing_extensions：

from typing import Callable, Awaitable, Optional, TypeVar, ParamSpec
from functools import wraps


_T = TypeVar('_T')
_P = ParamSpec('_P')

def catch_aiohttp_errors(func: Callable[_P, Awaitable[_T]]) -> Callable[_P, Awaitable[Optional[_T]]]:
    
    @wraps(func)
    async def wrapper(*args: _P.args, **kwargs: _P.kwargs) -> Optional[_T]:
        try:
            return await func(*args)
        except Exception:
            return None
    
    return wrapper
    

@catch_aiohttp_errors
async def download(url: str, download_path: str, filename: str, suffix: str) -> str:
    return 'foo'

现在download有签名

def (url: builtins.str, download_path: builtins.str, filename: builtins.str, suffix: builtins.str) -> typing.Awaitable[Union[builtins.str, None]]

此外，您现在不必手动添加 Optional - 装饰器即可。 Playground with this solution

Python 修饰异步函数的注释

Python annotations for decorated async functions

asynchronous

python-3.x

python-decorators

aiohttp

python-typing