Skip to content
Throttler for asyncio Python
Branch: master
Clone or download
Fetching latest commit…
Cannot retrieve the latest commit at this time.
Permalink
Type Name Latest commit message Commit time
Failed to load latest commit information.
.circleci
.coveragerc
LICENSE
README.md
aiothrottler.py
setup.py
test.py

README.md

aiothrottler CircleCI Test Coverage

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]

You can’t perform that action at this time.