Metadata-Version: 2.4
Name: aiodecorator
Version: 3.2.2
Summary: Binance Python SDK
Author-email: kaelzhang <i+github@kael.me>
Project-URL: Homepage, https://github.com/kaelzhang/python-aiodecorator
Classifier: Programming Language :: Python
Classifier: Programming Language :: Python :: 3.6
Classifier: Programming Language :: Python :: 3.7
Classifier: Programming Language :: Python :: 3.8
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Natural Language :: English
Classifier: Intended Audience :: Developers
Classifier: Operating System :: OS Independent
Classifier: Development Status :: 4 - Beta
Classifier: Topic :: Utilities
Requires-Python: >=3.6
Description-Content-Type: text/markdown
License-File: LICENSE
Provides-Extra: dev
Requires-Dist: codecov; extra == "dev"
Requires-Dist: coverage; extra == "dev"
Requires-Dist: ruff; extra == "dev"
Requires-Dist: pytest; extra == "dev"
Requires-Dist: pytest-asyncio; extra == "dev"
Requires-Dist: pytest-cov; extra == "dev"
Requires-Dist: twine; extra == "dev"
Requires-Dist: setuptools; extra == "dev"
Requires-Dist: mypy; extra == "dev"
Requires-Dist: build; extra == "dev"
Dynamic: license-file

[![](https://github.com/kaelzhang/python-aiodecorator/actions/workflows/python.yml/badge.svg)](https://github.com/kaelzhang/python-aiodecorator/actions/workflows/python.yml)
[![](https://codecov.io/gh/kaelzhang/python-aiodecorator/branch/master/graph/badge.svg)](https://codecov.io/gh/kaelzhang/python-aiodecorator)
[![](https://img.shields.io/pypi/v/aiodecorator.svg)](https://pypi.org/project/aiodecorator/)
[![](https://img.shields.io/pypi/l/aiodecorator.svg)](https://github.com/kaelzhang/python-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
<!-- - limit -->
<!-- - timeout -->

## Install

```sh
$ pip install aiodecorator
```

## Usage

```py
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:

```py
@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
```

```py
@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

```py
@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.

```py
@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](LICENSE)
