API

Task decorator

chard.task(fn: Callable[[...], Coroutine[Any, Any, Any]]) → TaskWrapper

Applying this decorator to an async function will turn it into a task object that will be discoverable by the Chard worker.

---

Apply it to any async function inside a Django apps’s tasks.py file like this:

import chard
import asyncio

@chard.task
async def my_task(i):
     print(i)
     await asyncio.sleep(1)

The resulting task object can be used to queue a new task for the worker like this:

my_task.send(123)

You can also call it synchronously:

await my_task(123)

Queueing a task returns a UUID that can be used to retrieve the corresponding Task model instance:

from chard.models import Task

task_id = my_task.send(123)
task = Task.objects.get(id=task_id)
task.status # => 'pending'

Task model

class chard.models.Task(id, status, created_at, updated_at, name, task_data)

class Status(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

All possible task status values.

DONE = 'done'

The task completed successfully.

FAILED = 'failed'

The task failed with an exception or it timed out.

PENDING = 'pending'

The task is queued.

RUNNING = 'running'

The task is running.

created_at

When the task was created.

id

Automatically assigned UUID.

name

The name of the task function e.g. myapp.tasks.my_task.

status

The status of the task.

task_data

JSON-encoded arguments for the task.

updated_at

When the task was last updated.

---

How to count the number of pending tasks:

from chard.models import Task

Task.objects.filter(status=Task.Status.PENDING).count() # => 5

Fetch a specific task:

from chard.models import Task

Task.objects.get(id="76bae380-d8b3-424e-8ce9-d70196e0fa7d")