Throttler for asyncio Python
Project description
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]
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Hashes for aiothrottler-0.0.16-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | f5f4c42c023c7f19619bd9331144ab060f34836be195b92e9281172345a294d7 |
|
MD5 | 3b2f3d7143cdf66c9fe9f28ca9ac6155 |
|
BLAKE2b-256 | a72d81f6a068085d25c392ed485fda4d8579ee3052c24e0ec47fc6cbb09d3a59 |