aiothrottler

aiothrottler

Throttler for asyncio Python

Installation

pip install aiothrottler

Usage

Create a shared Throttler, passing a minimum interval, e.g. 0.5 seconds

from aiothrottler import Throttler

throttler = Throttler(min_interval=0.5)

and then just before the piece(s) of code to be throttled, call this and await its result.

await throttler()
# There will be a gap of at least 0.5 seconds
# between executions reaching this line

Example: multiple tasks throttled

import asyncio
import time

from aiothrottler import Throttler

async def main():
    throttler = Throttler(min_interval=0.5)
    await asyncio.gather(*[
        worker(throttler) for _ in range(10)
    ])

async def worker(throttler):
    await throttler()
    # Interval of at least 0.5 seconds between prints
    # even though all workers started together
    print(time.time())

loop = asyncio.get_event_loop()
loop.run_until_complete(main())
loop.close()

Example: single task throttled/smoothed

import asyncio
import random
import time

from aiothrottler import Throttler

async def main():
    throttler = Throttler(min_interval=0.5)
    for _ in range(10):
        await throttler()
        # Interval of at least 0.5 seconds between prints
        # even though each sleep is random
        print(time.time())
        await asyncio.sleep(random.random())

loop = asyncio.get_event_loop()
loop.run_until_complete(main())
loop.close()

Differences to alternatives

• The API features a function to call to await its result [some use a context manager]

• The API is imperative [some use a functional approach/higher-order function]

• No polling is used [some use polling internally]

• A minimum interval between resolutions is used to throttle [rather that a max resolutions per time interval, which can cause an irregular pattern of resolutions]

• The tests cover edge cases, such as asserting on throttling after tasks being throttled have been cancelled [some alternatives do not]