Declaring Tasks

Tasks are what you'll be scheduling and what would be executed by the worker.

To declare a task you simply need to decorate a function with task decorator:

from aiotaskqueue import task


@task(name="task-name")
async def my_task() -> None:
    pass

TaskRouter

Task router is a simpler way to create a collection of tasks:

from aiotaskqueue import TaskRouter

router = TaskRouter()


@router.task(name="task-name")
async def my_task() -> None:
    pass

Markers

Different extensions could provide markers to add metadata to your task definitions, for example to add retries to your tasks you could use Retry marker:

from aiotaskqueue import TaskRouter
from aiotaskqueue.extensions.builtin import Retry

router = TaskRouter()


@router.task(
    name="task-name",
    markers=[Retry(max_retries=3)],
)
async def my_task() -> None:
    pass

Retry Extension

In order for this specific example to work you need to add RetryExtension to your Configuration