Абстрактный метод Python с asynccontextmanager

Question

У меня есть пара классов ограничения скорости (один не показан), для которых я хотел бы создать азбуку. Метод request - это асинхронный менеджер контекста. С кодом, показанным ниже, я получаю

Подпись "request" несовместима с супертипом "RateLimiterInterface"

Если я попытаюсь украсить абстрактный метод с помощью @asynccontextmanager, я получу ошибку ввода:

Аргумент 1 для «asynccontextmanager» имеет несовместимый тип «Callable [[RateLimiterInterface], Coroutine [Any, Any, AsyncIterator [Any]]]»; ожидается "Callable [..., 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

Я не могу воспроизвести проблему. Код ниже работает хорошо для меня (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())