带有异步上下文管理器的 Python 抽象方法

Question

我有几个速率限制器类（一个未显示），我想为其创建一个 ABC。 request 方法是一个异步上下文管理器。使用下面显示的代码，我得到

“请求”的签名与超类型“RateLimiterInterface”不兼容

如果我尝试用@asynccontextmanager 来装饰抽象方法，我会收到输入错误：

“asynccontextmanager”的参数 1 具有不兼容的类型“Callable[[RateLimiterInterface], Coroutine[Any, Any, AsyncIterator[Any]]]”；预期“可调用[...，AsyncIterator[]]”

我该怎么做？

class RateLimiterInterface(abc.ABC):
    @abc.abstractmethod
    async def request(self) -> AsyncIterator:
        pass

class LeakyBucketRateLimiter(RateLimiterInterface):
    def __init__(self, max_tokens: Optional[int] = None, rate: float = 60) -> None:
        self.max_tokens = max_tokens
        self.rate = rate
        self._bucket = max_tokens
        self._last_added_at = time.time()

    @contextlib.asynccontextmanager
    async def request(self) -> AsyncIterator:
        if self._bucket is None:
            yield
            return
        while not self._bucket:
            await asyncio.sleep(0)
            self._add_tokens(int((time.time() - self._last_added_at) * self.rate))
        self._bucket -= 1
        yield
        return

    def _add_tokens(self, num_tokens: int) -> None:
        if num_tokens == 0:
            return
        self._bucket += num_tokens
        if self._bucket > self.max_tokens:
            self._bucket = self.max_tokens
        self._last_added_at = time.time()

Answer 1

我刚刚在打字时遇到了同样的问题并以这种方式解决了：

import abc
import contextlib
import asyncio


class TestAbstract(metaclass=abc.ABCMeta):
    @contextlib.asynccontextmanager
    @abc.abstractmethod
    # Here the trick: you must declare an  asynchronous generator function,
    # not a regular coroutine function so it have to explicitly yield.
    async def foo(self):
        yield


class SubTest(TestAbstract):
    @contextlib.asynccontextmanager
    async def foo(self):
        yield


async def test_coro():
    async with SubTest().foo():
        print('OK')


asyncio.run(test_coro())

Answer 2

我无法重现问题。下面的代码对我来说运行良好（Python 3.7.1）。

请测试它是否在您的环境中运行良好，以及是否提供最小的可重现示例。

import asyncio
import abc
import contextlib
from typing import AsyncIterator


class Test(abc.ABC):
    @abc.abstractmethod
    async def request(self) -> AsyncIterator:
        pass


class SubTest(Test):
    @contextlib.asynccontextmanager
    async def request(self) -> AsyncIterator:
        await asyncio.sleep(1)
        yield 1
        await asyncio.sleep(1)


async def main():
    obj = SubTest()
    async with obj.request() as res:
        print(res)


asyncio.run(main())