rate_limit.py

#### work

import typing as t
import asyncio


RateResult = t.Callable[..., t.Coroutine[t.Any, t.Any, None]]


def rate_limit_builder(count_limit: int, time_period: int) -> RateResult:
    loop = asyncio.get_event_loop()
    counter = {"count": 0, "fut": loop.create_future()}

    async def clean_rate_limit() -> None:
        while True:
            await asyncio.sleep(time_period)
            if not counter['fut'].done():
                counter['fut'].set_result(True)
                counter["fut"] = loop.create_future()

    asyncio.create_task(clean_rate_limit())

    async def inner_call() -> None:
        counter["count"] += 1

        if counter["count"] == count_limit:
            await counter["fut"]
            counter["count"] = 0

    return inner_call


async def test():
    limit = rate_limit_builder(10, 1)

    for i in range(410):
        print(i)
        await limit()
        await asyncio.sleep(0.2)

asyncio.run(test())

### not work

import asyncio


def rate_limit_builder(max_count: int, time: int):
    lock = asyncio.Lock()

    counter = {"count": max_count}

    async def clean_rate_limit(inner_lock):
        while True:
            await asyncio.sleep(time)
            if inner_lock.locked():
                await inner_lock.release()

    asyncio.create_task(clean_rate_limit(lock))

    async def inner_call(inner_lock):
        counter["count"] -= 1

        if counter["count"] == 0:
            await inner_lock.acquire()
            counter["count"] = max_count

    return lambda: inner_call(lock)


async def test():
    limit = rate_limit_builder(10, 1)
    for i in range(40):
        print(i)
        await asyncio.sleep(0.1)
        await limit()


asyncio.run(test())