Skip to main content

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.

Source Distribution

aiothrottler-0.0.16.tar.gz (2.4 kB view hashes)

Uploaded Source

Built Distribution

aiothrottler-0.0.16-py3-none-any.whl (3.4 kB view hashes)

Uploaded Python 3

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page