aiodecorator

aiodecorator

Python decorators for asyncio, including

• timeout: Set the timeout for a function
• throttle: Throttle a (coroutine) function that return an Awaitable
• repeat: Repeat a function
• schedule_naturally: Schedule a function to run from the next time moment

Install

$ pip install aiodecorator

Usage

import time
import asyncio

from aiodecorator import (
    throttle
)


async def run(throttle_type):
    now = time.time()

    # -----------------------------------------------------
    # The throttled function is only called twice a second
    @throttle(2, 1, throttle_type)
    async def throttled(index: int):
        await asyncio.sleep(0.1)
        diff = format(time.time() - now, '.0f')
        print(index, f'{diff}s')
    # -----------------------------------------------------

    loop = asyncio.get_running_loop()
    tasks = [
        loop.create_task(throttled(index))
        for index in range(5)
    ]

    await asyncio.wait(tasks)


asyncio.run(run('wait'))

# Output: queued
# 0 0s
# 1 0s
# 2 1s
# 3 1s
# 4 2s

asyncio.run(run('ignore'))

# Output: 2, 3, 4 are ignored
# 0 0s
# 1 0s


asyncio.run(run('replace'))

# Output: 1, 2, 3 are canceled
# 0 0s
# 4 0s

APIs

throttle(limit: int, interval: Union[float, int], throttle_type)

• limit int Maximum number of calls within an interval.
• interval Union[int, float] Timespan for limit in seconds.
• throttle_type Literal['ignore', 'wait', 'replace'] = 'ignore'
  • 'ignore': ignore the function call and return None if it exceeds the limit.
  • 'wait': wait for the next tick to execute the function.
  • 'replace': try to cancel the last function call, let it return None and execute the current function call. (New in 3.1.0)

Returns a decorator function

schedule_naturally(unit, delay, weekday)

• unit Literal['secondly', 'minutely', 'hourly', 'daily', 'weekly', 'monthly', 'yearly']
• delay timedelta = timedelta(seconds=0)
• weekday Weekday only used when unit is 'weekly'

Returns a decorator function that schedule a function fn to run from the next time moment with a delay delay

For example:

@schedule_natually('daily', delay = timedelta(seconds=50))
async def run():
    print('hello')

await run()

# It will print 'hello' at 00:00:50 in the next day

@repeat(-1)
@schedule_naturally(
    'weekly',
    delay = timedelta(minutes=5),
    weekday = 'Wednesday'
)
async def run_weekly():
    print('hello')

await run_weekly()

# It will print 'hello' at 00:05 every Wednesday

repeat(times: int, interval: float = 0.)

• times int the number of times to repeat the function
• interval float = 0. the interval between each call

@repeat(7)
@schedule_naturally('daily')
@repeat(3, 0.1)
async def run()
    print('hello')

await run()

# It will schedule a one-week plan, at 00:00:00 each day, it prints "hello" three times, with 100 ms between each print.

timeout(seconds: int | None, at: float | None)

New in 3.1.0

• seconds int | None = None seconds to time out. If seconds <= 0 or seconds is None, there will be no timeout.
• at (float | None = None): deadline to timeout, at has higher priority than seconds

Make the function automatically cancel itself if it takes longer than seconds seconds.

@timeout(1)
async def my_function():
    await asyncio.sleep(2)

try:
    await my_function():
except asyncio.TimeoutError:
    print('timeout')

# It will print 'timeout'

License

MIT