concurrent-tasks 1.1.0

Tools to run asyncio tasks concurrently.

asyncio-concurrent-tasks

Tooling to run asyncio tasks.

• Background task
• Threaded task pool
• Restartable task

Background task

Task that is running in the background until cancelled. Can be used as a context manager.

Example usage:

import asyncio
from typing import Callable, Awaitable
from concurrent_tasks import BackgroundTask


class HeartBeat(BackgroundTask):
    def __init__(self, interval: float, func: Callable[[], None]):
        super().__init__(self._run, interval, func)

    async def _run(self, interval: float, func: Callable[[], Awaitable]) -> None:
        while True:
            await func()
            await asyncio.sleep(interval)

Threaded task pool

Run async tasks in a dedicated thread. It will have its own event loop.

Parameters:

• name will be used as the thread's name.
• size can be a positive integer to limit the number of tasks concurrently running.
• timeout can be set to define a maximum running time for each time after which it will be cancelled. Note: this excludes time spent waiting to be started (time spent in the buffer).
• context_manager can be optional context managers that will be entered when the loop has started and exited before the loop is stopped.

💡 All tasks will be completed when the pool is stopped.

💡 Blocking and async version are the same, prefer the async version if client code is async.

Blocking

This can be used to run async functions in a dedicated event loop, while keeping it running to handle background tasks

Example usage:

from concurrent_tasks import BlockingThreadedTaskPool


async def func():
    ...


with BlockingThreadedTaskPool() as pool:
    # Create and run the task.
    result = pool.run(func())
    # Create a task, the future will hold information about completion.
    future = pool.create_task(func())

Async

Threads can be useful in cooperation with asyncio to let the OS guarantee fair resource distribution between threads. This is especially useful in case you cannot know if called code will properly cooperate with the event loop.

Example usage:

from concurrent_tasks import AsyncThreadedTaskPool


async def func():
    ...


async with AsyncThreadedTaskPool() as pool:
    # Create and run the task.
    result = await pool.run(func())
    # Create a task, the future will hold information about completion.
    future = pool.create_task(func())

Restartable task

Task that can be started and cancelled multiple times until it can finally be completed. This is useful to handle pauses and retries when handling with a connection.

💡 Use functools.partial to pass parameters to the function.

Example usage:

from functools import partial
from concurrent_tasks import RestartableTask

async def send(data): ...

task: RestartableTask[int] = RestartableTask(partial(send, b"\x00"), timeout=1)
task.start()
assert await task == 1

# Running in other tasks:

# On connection lost:
task.cancel()
# On connection resumed:
task.start()
# On response received:
task.set_result(1)