如何为支持同步和异步接口的库生成 python 类型信息？

Question

我在 Python 中的输入和类型提示方面都遇到问题。我准备了一个（可执行）示例，展示了我所面临的应该公开同步和异步接口的库的问题。这显然是一个简化的示例，消除了噪音并专注于当前的问题：

import asyncio
import time

class CommonFunctions:
    # I want to use this as the interface defintion
    # for my async and sync implementations below
    
    def do_something(self, i: int, text: str = "hello world") -> str:
        return text * i
    
    def do_something_else(self, i: int, j: int) -> str:
        return i * j

    # Imagine more functions here... This is 
    # the library and where most development happens.


class SyncLib(CommonFunctions):
    def __complex_operation(self):
        # This is a standin for sync-api calls
        time.sleep(1)
    
    def do_something(self, *args, **kwargs):
        self.__complex_operation()
        return super().do_something(*args, **kwargs)
    
    def do_something_else(self, *args, **kwargs):
        self.__complex_operation()
        return super().do_something_else(*args, **kwargs)


class AsyncLib(CommonFunctions):
    async def __complex_async_operation(self):
        # This is a standin for async-api calls
        await asyncio.sleep(1)
    
    async def do_something(self, *args, **kwargs):
        await self.__complex_async_operation()
        return super().do_something(*args, **kwargs)
    
    async def do_something_else(self, *args, **kwargs):
        await self.__complex_async_operation()
        return super().do_something_else(*args, **kwargs)


print("User using the sync-lib...")
sync_lib = SyncLib()
print(sync_lib.do_something(5))
print(sync_lib.do_something_else(5, 6))

print("User using the async-lib...")
async_lib = AsyncLib()
async def async_main():
    print(await async_lib.do_something(5))
    print(await async_lib.do_something_else(5, 6))
asyncio.get_event_loop().run_until_complete(async_main())

类SyncLib和AsyncLib是库用户将使用的接口。对于不同的用例，库应该具有由类提供的同步和异步接口SyncLib（AsyncLib对于感兴趣的人：实际的库基于 httpx 同步/异步客户端，这里用调用表示time.sleep()/asyncio.sleep()）。

大多数库实际上是同步的，但对于一小部分，它取决于对底层库的调用，这里用调用来表示__complex_operation。

问题

对于该库的用户来说，这种设计的问题在于 AsyncLib（和 SyncLib）的键入信息和类型提示完全不正常。所有公共成员函数看起来都接受*args, **kwargs参数。我可以通过为 CommonFunctions 类添加一个协议来轻松解决这个问题，该协议准确定义了每个函数的可用参数和参数类型。

然而，这样的协议将仅定义同步调用。异步调用需要对其自己的协议进行第二次声明，该协议只是原始同步协议的副本，所有函数声明都被标记为异步。

突然间，如果需要向函数添加参数，开发人员需要维护三个接口：CommonFunctions、同步协议和异步协议（加上文档字符串）。

一定会有更好的办法？！我的想法和问题：有没有一种方法可以从同步协议生成异步协议，以便 IDE 中的类型库和/或类型提示工具可以理解它们？

我尝试过的...

好吧，我读了一堆 PEP 等，但实际上没有找到很多用于将一组签名从一个协议转换为另一个协议的内容。即使将单个函数的签名从同步更改为异步，同时以通用方式保留同步函数的参数列表似乎也很复杂。

对于单一功能的情况，我正在考虑类似装饰器的东西，它可以执行以下操作：

import typing

Result = typing.TypeVar("Result")
Args = typing.TypeVar("Args")

def typing_sync_to_async(func: typing.Callable[Args, Result]) -> typting.Callable[Result, typing.Coroutine[typing.Any, typing.Any, Result]]:
    return func

@typing_sync_to_async
def sync_example(x: int) -> str:
    pass

但是，这行不通，因为

1. 参数列表不能像这样捕获
2. **kwargs正如我发现的那样，这并不包括在内，因为Callable它似乎没有实现定义关键字参数类型的方法。

对我来说，做到这一点的唯一方法似乎是手动为数百个函数定义这些协议（异步和同步），并且我不能依赖提供此功能的现有功能。那是对的吗？也许，如果我更多地了解打字库并自己贡献一些东西？

Answer 1

Paramspec这可以使用引入来解决PEP-612

这是您的示例中修改后的装饰器：

from typing import ParamSpec, TypeVar
from collections.abc import Awaitable, Callable

P = ParamSpec("P")
R = TypeVar("R")

def to_async(func: Callable[P, R]) -> Callable[P, Awaitable[R]]:
    ...

这将捕获必要的函数签名，包括关键字参数。

演示正确类型推断的示例：