ratelimiter_lite 0.1.4

Rate Limit Components

version: 0.1.4

pip install ratelimiter_lite

Demo Usage:

import time
from concurrent.futures.thread import ThreadPoolExecutor

from ratelimiter_lite.factory import ratelimiter_lite_factory
from ratelimiter_lite.fixed_window import FixedWindowratelimiter


@ratelimiter_factory(FixedWindowratelimiter, limit=2, period=0.3)
def task(args):
   print(f"task: {args}")


def test_fixed_window():
   # limit worker count，observe ratelimiter_lite functions.
   with ThreadPoolExecutor(max_workers=5) as p:
      period = 2.0
      start = time.monotonic()
      while True:
            if time.monotonic() - start > period:
               break

            p.submit(task, "running.")
            # wait thread for a while, the period should less than limiter's `period/limit`
            time.sleep(0.05)