Worker

A worker consume tasks. It register coroutines as tasks, under a task_name. It use by default the coroutine name - but be carefull of decorators because of the __name__ attribute.

Like the broker, worker rely on a queue and a cache. Check the broker page for available queues and caches.

It’s better to setup a worker first (before the broker), so you’re sure the queue is setup.

Setup a worker

from aio_task import Worker

worker = await Worker.create(
    queue_type="rabbitmq",  # desired type of queue
    queue_conf={"url": "amqp://guest:guest@localhost:5672",
                "routing_key": "tasks_queue"},  # dict with the queue configuration - depends on the queue type.
    cache_type="redis",  # desired type of cache
    cache_conf={"address": "redis://localhost"},  # dict with the cache configuration - depends on the cache type.
)

These params should match with the broker ones.

Register tasks

async def coro(**kwargs):  # task example
  pass

async def coro2(**kwargs):
  pass

worker.register_handler(
    coro=coro,  # the coroutine corresponding to the task.
    task_name="coro"  # optional → use coro.__name__
)
worker.register_handler(coro2)  # register another task
worker.register_handler(coro2, task_name="task1")  # register task under another name

When you have register all your tasks, start the worker to consume tasks.

await worker.start()

Once the worker is started, you cannot register other tasks.

Error management

If an exception occure while processing the task, the exception will be logged by the worker and set as task’s result.

async def coro():  # example of task giving an error
    raise ValueError("Oops")

# setup and start the worker
worker.register_handler(coro)
await worker.start()
# Then, from the broker perspective
task_id = await broker.create_task("coro")
await asyncio.sleep(0.1)  # time for task to be processed
task = await broker.get_task(task_id)  # fetch task from the cache

assert task.result == {"exception": {"class": "ValueError",
                                     "args": ("Oops",),
                                     "str": "Oops"}}

The task result has to be serializable (with json.dumps), otherwise worker will log an error and set the error as task’s result.

# task result if the result is not serializable.
{'exception': {'args': ('Task result is not serializable',),
               'class': 'ValueError',
               'str': 'Task result is not serializable'}}

Shutdown

await worker.close()

That’s it!