Skip to main content
Donate to the Python Software Foundation or Purchase a PyCharm License to Benefit the PSF! Donate Now

Throttler for asyncio Python

Project description

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]

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Filename, size & hash SHA256 hash help File type Python version Upload date
aiothrottler-0.0.16-py3-none-any.whl (3.4 kB) Copy SHA256 hash SHA256 Wheel py3
aiothrottler-0.0.16.tar.gz (2.4 kB) Copy SHA256 hash SHA256 Source None

Supported by

Elastic Elastic Search Pingdom Pingdom Monitoring Google Google BigQuery Sentry Sentry Error logging AWS AWS Cloud computing DataDog DataDog Monitoring Fastly Fastly CDN SignalFx SignalFx Supporter DigiCert DigiCert EV certificate StatusPage StatusPage Status page