asynctoolspy 1.4

asynctoolspy is a set of simple tools to speed up Python async development. It includes the appoint_limit_async decorator to limit how often an async function can be called by setting max attempts and intervals, useful for APIs with rate limits. The decorator handles errors with retries and pauses to improve reliability. It also features AsyncIterWrapper, an async iterator supporting optional delays and sync/async callbacks for flexible data processing.

asynctoolspy

Asynctoolspy is a collection of simple tools to speed up Python async development.
This project contains a appoint_limit_async decorator that limits the frequency of calls to the function to which it is applied.
The decorator allows you to set the maximum number of attempts to call a function and the time interval between them.
This is useful when the API has restrictions on the number of requests per unit of time.\ The decorator is race-condition safe in an event loop environment.
The decorator wraps the function call in a try-except block, which allows you to handle possible exceptions that occur when executing the request.
This can be useful when dealing with unreliable or temporarily unavailable APIs, where retries can help ensure the request succeeds.If an error occurs and the number of attempts does not exceed the set limit, the decorator pauses and re-calls the function.
Now features AsyncIterWrapper an async iterator for any iterable, supporting optional delays and sync/async callbacks for flexible data processing. The @run_in_background decorator allows you to trigger an asynchronous function and continue execution immediately. It is specifically designed for tasks that do not require their results to be awaited, such as:

• Timers and countdowns in Telegram bots.
• Notifications and logging.
• Background cleanup or data synchronization.

The @run_in_background decorator prevents the event loop from being blocked, ensuring your application remains responsive.

Requirements

Python >=3.8.

Install

pip install asynctoolspy or uv pip install asynctoolspy or uv add asynctoolspy

Usage example №1

from asynctoolspy import appoint_limit_async


class MyClient:
    @appoint_limit_async(limit=5, interval=10, pause=2)
    async def my_method(self):
          # Your code for making an API request or I/O operation.

@appoint_limit_async(limit=5, interval=10, pause=2)
async def my_function():
      # your code for making I/O operation

Usage example №2

import asyncio 
import httpx 
import sys
from asynctoolspy import AsyncIterWrapper


async def check_url(url):
    methods = AsyncIterWrapper(['GET', 'HEAD', 'OPTIONS', 'POST', 'PUT', 'DELETE', 'PATCH'])
    available_methods = {}

    async with httpx.AsyncClient() as client:
        async for method in methods:
            try:
                response = await client.request(method, url)
                if response.status_code != 405:
                    available_methods[method] = response.status_code
            except httpx.RequestError:
                continue

    return available_methods

async def main(urls):
    results = {}
    async for url in urls:
        if url.startswith(('http://', 'https://')):
            results[url] = await check_url(url)
        else:
            print(f'String "{url}" is not url.')

    print(results)

if __name__ == "__main__":

    with open('data_link.txt', 'r') as file:
        lines = [line.strip() for line in file]
    urls_data = AsyncList(lines)
    asyncio.run(main(urls_data))

Usage example №3

# This is an example with in the aiogram 3.x

....
....
from asynctoolspy import AsyncIterWrapper
from asynctoolspy import run_in_background
...

user_router = Router()

@user_router.message(CommandStart())
@run_in_background
async def cmd_start(message: Message, state: FSMContext, bot: Bot) -> None:
    await message.answer(
        f"Hello, \n\n"
    )
    await state.set_state(UserStates.GET_TASK)
    task_message = await message.answer(text=f'time: - s.\nWhat is X equal to?: 10 + 5 + 8 = Х')
    async for sec in AsyncIterWrapper(data=[7, 6, 5, 4, 3, 2, 1, 0], delay=1):
        if await state.get_state() != UserStates.GET_TASK:
            return
        try:
            await bot.edit_message_text(
                chat_id=message.chat.id,
                message_id=task_message.message_id,
                text=f'time: {sec} s.\nWhat is X equal to?: 10 + 5 + 8 = Х'
            )
        except Exception:
            pass
    await state.clear()
    await message.answer('Game over!')


@user_router.message(F.text.regexp(r'^-?\d+(\.\d+)?$'), UserStates.GET_TASK)
async def resolve_task(message: Message, state: FSMContext, bot: Bot) -> None:
    await state.clear()
    result = 'correct' if int(message.text)==23 else 'not true'
    await message.answer(text=f'Thank you.\n{result}.\nYour decision: {message.text}')

Usage example №4

from asynctoolspy import AsyncIterWrapper

async def multiply(x):
     await asyncio.sleep(0.1)
     return x * 2

data = [1, 2, 3]

async for item in AsyncIterWrapper(data, delay=0.5, callback=multiply):
     print(item)  # Outputs 2, 4, 6 with a 0.5 second delay between items