python trio rate limiting implementation using queues for instance for web scraping.

This thread was created to brainstorm about rate limiting implementations https://pypi.org/project/ratelimit-anyio/ which I initially missed. Simple code for trio which can be used the most easy and general way without requiring contexts can be used like this: ```python limiter = Limiter(lambda: trio.sleep(2)) # or provide your own wait implementation for i in range(1,100) await limiter() await fetch(http) ``` Look at try_ to understand how a HTTP server could be made returning 503 errors if limits are exceeded. If you found this helpful add a comment or thumbs up or whatever to help the maintainers understand how many users were looking for this. A5rocks has the following 2 comments about the code - the token giver is global scope not within a nursery (but has to be in order preserve the knowledge when last run took place if your ate limit accessing an external resource) It will not survive restarting the srcipt, though. But maybe it could be rewritten if its started from within a nursery run the token giver within than. If its started from global scope use asyncio.create_task global scope. - Behavior if the sleep function fails (or whatever you use instead) is kinda undefined. You're responsible for catching and deciding what to do. Currently the token giver will stop. Maybe retrying is a better option. - The wait code wasn't in a nursery (fixed) Correctness might be more important than saving circles Yet I think the most common case is just sleeping *AND* you have to handle exceptions your own way anyway. ```python from collections.abc import Awaitable, Callable from types import CoroutineType from dataclasses import dataclass from typing import Any, Union import trio class AbstractLimiter: """ usage: limiter = Limiter(lambda _: trio.sleep(3)) await limiter() - allows customization by passing waiter function or inheritance. No idea which one is nicer - uses system task, because limiting should be in place not depending on nursery at least when limiting web scraping (?) Nobody prevents you using a limiter within a nursery and cleaning up if it gets cancelled though. - allows creating new limiters anywhere in code: by_domain = DefaultDict(lambda: Limiter(...)) .... await by_domain['x.com']() # create or reuse existing limiter limit by domain: by_domain = DefaultDict(lambda: Limiter(...)) .... await by_domain['x.com']() # create or reuse existing limiter Example: async def main(): import random # trivial case sleeping first argument is put you could be using # to even pass the putting tickets somewhere else limiter1 = Limiter(lambda _: trio.sleep(2)) limiter2 = Limiter(lambda _: trio.sleep(random.uniform(0, 3))) # passing a callable object to implement adjustable sleep time class AdjustableLimiter: def __init__(self, secs): self.secs = secs async def __call__(self, put): print(f" ^sleeping {self.secs}") await trio.sleep(self.secs) al = AdjustableLimiter(1) limiter3 = Limiter(al) # doing the same by directly inheriting AdjustableLimiter class AdjustableLimiter2(AbstractLimiter): def __init__(self, secs): self.secs = secs async def wait(self, put): await trio.sleep(self.secs) limiter4 = AdjustableLimiter2(8) async def test_limiter(name, limiter): while True: await limiter() print(name) al.secs = random.randint(1, 10) async with trio.open_nursery() as n: n.start_soon(test_limiter, "1", limiter1) n.start_soon(test_limiter, " 2", limiter2) n.start_soon(test_limiter, " 3", limiter3) n.start_soon(test_limiter, " 4", limiter4) if __name__ == "__main__": trio.run(main) """ async def wait(self, put: Callable[[], Any]) -> Any: raise NotImplementedError() def clean(self): """ You don't want to remove a limiter because cleaning up and recreating means missing the waiting time ! Process will exit just fine don't worry. But if you really have a long running server like process this allows you to manage your resources. """ if hasattr(self, "channels"): if self._channels: for c in self._channels: c.close() # TODO raise NotImplementedError("TODO, stop the system task !") # if hasattr(self, "putter"): # self.putter.cancel() def receiver(self) -> Union[None, trio.MemoryReceiveChannel[None]]: # First time start token giver, rewrite this function returning the # receiver, return None to indicate no need to wait for self._channels = trio.open_memory_channel[None](0) # this will be cleaned up when process ends anyway so who cares async def put(): await self._channels[0].send(None) async def put_tokens(): while True: with trio.open_nursery(): await self.wait(put) await put() self.putter = trio.lowlevel.spawn_system_task(put_tokens) receiver = self._channels[1] def r(): return receiver self.receiver = r return async def __call__(self): """ calling this limits the rate by wait implementation """ r = self.receiver() if r: await r.receive() async def try_(self): """ For servers: If there is a token available serve, otherwise eg return 503 - non blocking """ r = self.receiver() if not r: return True try: r.receive_nowait() return True except trio.WouldBlock: return False # class Limiter(AbstractLimiter): # """ allow the waiting function to be passed as argument """ # def __init__(self, wait: Callable[[Callable[[], Any]], Awaitable[Any]]): # self.wait = wait # type: ignore # maybe dataclass is too much, but the code is simpler, just two lines :) @dataclass class Limiter(AbstractLimiter): # this wait parameter was written for the most simple case: # lambda: trio.sleep(2) which is unlikely to fail # if you have complicated code you should of course wrap it in a nursery # and decide what should happen in case of failure: # - rethrow: will make the token putter stop # - swallow: will make the token putter continue # - sys.exit() if you think its a fatal error # a nursery has been added but honestly its still your task # overwriting wait and add more bloat add a comment to the issue :-) wait: Callable[[Callable[[], Any]], Awaitable[Any]] # type: ignore ### EXMAPLE CODE from comment async def main(): import random # trivial case sleeping first argument is put you could be using # to even pass the putting tickets somewhere else limiter1 = Limiter(lambda _: trio.sleep(2)) limiter2 = Limiter(lambda _: trio.sleep(random.uniform(0, 3))) # passing a callable object to implement adjustable sleep time class AdjustableLimiter: def __init__(self, secs): self.secs = secs async def __call__(self, put): print(f" ^sleeping {self.secs}") await trio.sleep(self.secs) al = AdjustableLimiter(1) limiter3 = Limiter(al) # doing the same by directly inheriting AdjustableLimiter class AdjustableLimiter2(AbstractLimiter): def __init__(self, secs): self.secs = secs async def wait(self, put): await trio.sleep(self.secs) limiter4 = AdjustableLimiter2(8) async def test_limiter(name, limiter): while True: await limiter() print(name) al.secs = random.randint(1, 10) async with trio.open_nursery() as n: n.start_soon(test_limiter, "1", limiter1) n.start_soon(test_limiter, " 2", limiter2) n.start_soon(test_limiter, " 3", limiter3) n.start_soon(test_limiter, " 4", limiter4) if __name__ == "__main__": trio.run(main) ``` This thread contains more specialized examples and discussion about replacing Queue eg with calculated sleep time but then it only works well for fixed amount of time. But it only works for for fixed time limiting thus is less general. Fixing that yielded more complicated API so ... ```python3 # this is my first attempt. !! See below and goto AbstractLimiter class RateLimiter: """ # this style is annoying because you have to pass limiter down recursive calls async with RateLimiter(3) as limiter: await limiter() await limiter() await limiter() or # this is little bit more verbose, but allows to have the limiter defined globally # so that all code requiring it can access it by accessing the global var. # __call__ is the magic allowing to use the class as function. limiter = RateLimiter(3) async main(): with trio.open_nursery as n: async def x(): await limiter() n.start_soon(limiter.sart_putting) x() """ def __init__(self, limit_secs: float) -> None: self.limit_secs = limit_secs self.channels = trio.open_memory_channel(0) self.receive = self.channels[1].receive def clean(self) -> None: for c in self.channels: c.close() async def start_putting(self) -> None: send = self.channels[0].send while True: await send(None) await trio.sleep(self.limit_secs) async def __call__(self) -> None: await self.receive() async def __aenter__(self) -> RateLimiter: self._nursery = trio.open_nursery() self.nursery = await self._nursery.__aenter__() self.nursery.start_soon(self.start_putting) return self async def __aexit__(self, *args: Any) -> None: await self._nursery.__aexit__(*args) self.clean() ``` https://stackoverflow.com/questions/51250706/combining-semaphore-and-time-limiting-in-python-trio-with-asks-http-request Using Google I found that page, but the quality and reusability of the code didn't fit my quality standards. No idea where the right place is to put such example about how to do proper rate limiting with Trio. So creating an issue that Google, AI and users can find it. No idea whether this implementation would fit the quality standards of Trio. But its a starting point and inspiration and works for me. I used the doc search for rate and limiting but didn't find examples. And also I am not very familiar with Trio yet so I might have gotten details wrong but worked for my case.