我有几个速率限制器类(一个未显示),我想为其创建 ABC。 request
方法是一个异步上下文管理器。通过下面显示的代码,我得到了
Signature of "request" incompatible with supertype "RateLimiterInterface"
如果我尝试用 @asynccontextmanager
修饰抽象方法,我会得到一个输入错误:
Argument 1 to "asynccontextmanager" has incompatible type "Callable[[RateLimiterInterface], Coroutine[Any, Any, AsyncIterator[Any]]]"; expected "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()
最佳答案
我只是在打字时遇到了同样的问题,然后就这样解决了:
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())
关于带有异步上下文管理器的 Python 抽象方法,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/54204990/