kapusta 0.1.3

A simple asynchronous task manager using sqlalchemy as a broker and result backend

Kapusta

Kapusta is a simple asynchronous task manager that allows using SQLAlchemy as a broker and result backend (with the possibility of extension). It can work both in the main thread and scale up to multiple processes. It is designed for low-load systems where the number of tasks is small but critical.

[!WARNING]

This task manager is intended for use cases where the number of tasks is so small that there is no point in creating additional processes and integrating third-party tools (e.g., running Celery, Celery-beat, RabbitMQ for a few deferred tasks per hour).

Installation

To install Kapusta, clone the repository and install the required dependencies:

pip install kapusta

Simple Example

import asyncio

from kapusta import AlchemyCRUD, Kapusta

DB_URL = '...'

kapusta = Kapusta(
    crud=AlchemyCRUD(DB_URL)
)


@kapusta.task()
def task():
    print('test')


async def main():
    await kapusta.startup()  # At the start of the application.

    await task.launch()
    # in this example, you need to wait for the task to be completed in the main thread.
    await asyncio.sleep(1)

    await kapusta.shutdown()  # At the end of the application


asyncio.run(main())

Usage

Running a Kapusta Application

kapusta = Kapusta(
    crud=AlchemyCRUD(DB_URL),
    logger=Logger('kapusta'),
    max_tick_interval=60 * 5,
    default_overdue_time_delta=None,
    default_max_retry_attempts=0,
    default_timeout=60
)

When initializing Kapusta, the following arguments are accepted:

• crud: An object inherited from BaseCRUD. Implements the logic of the broker and result backend.
• logger: An object from the standard logging module.
• max_tick_interval: Defines the maximum time between checks for pending tasks that should be executed soon.
• default_overdue_time_delta: Passed by default when registering a task if not overridden. A timedelta with the specified time when the task will be considered overdue, None if the task should be executed in any case.
• default_max_retry_attempts: Passed by default when registering a task if not overridden.
• default_timeout: Passed by default when registering a task if not overridden.

await kapusta.startup()

When the application starts, an asyncio.create_task method kapusta.listening is created, which in an infinite loop asynchronously accesses the CRUD service (in the classic implementation, this is AlchemyCRUD), takes tasks from there, and launches them. Also, crud.startup() is called, necessary for setting up the result backend and broker.

Defining Tasks

A task is a synchronous or asynchronous function. To register a function as a task, you can use the kapusta.register_task method or the @kapusta.task decorator (@kapusta.task internally calls kapusta.register_task).

# Using the decorator

@kapusta.task()
def sample_task(param1, param2):
    return param1 + param2

# Using the method

def sample_task(param1, param2):
    return param1 + param2

kapusta.register_task(sample_task)

When using both methods, you can pass the following arguments:

"""
execution_mode (TaskExecutionMode, optional): The execution mode of the task
    (sync, async_, thread, or process). Defaults to Sentinel.
eta_delta (timedelta, optional): The time delta for the estimated time of
    execution. If None, the task is executed immediately. Defaults to None.
overdue_time (timedelta): Time at which a task will be considered overdue.
    Defaults to None.
max_retry_attempts (int, optional): The maximum number of retry attempts
    for the task. Defaults to Sentinel.
timeout (Seconds, optional): The maximum time allowed for task execution.
    Defaults to Sentinel.
"""

Launching Tasks

You can launch tasks using the launch or launch_now methods.

await sample_task.launch(param1=1, param2=2)
await sample_task.launch_now(param1=1, param2=2)

At this point, you can also change the task parameters by passing them in update_params:

await sample_task.launch(param1=1, param2=2, update_params={'max_retry_attempts': 1})

Retrieving Task Results and Status

You can retrieve the result of a completed task using the get_task_result method.

result = await kapusta.get_task_result(task_id)
status = await kapusta.get_task_status(task_id)

CRUD Interface

Kapusta uses a CRUD interface for database operations related to tasks. You can implement your own CRUD interface by extending the CRUDProtocol class.

from kapusta import BaseCRUD

class MyCRUD(BaseCRUD):
    # Implement the required methods
    ...

Testing

Kapusta uses pytest for testing. You can run the tests using the following command:

pytest