Description

# Django Tasks [](https://github.com/RealOrangeOne/django-tasks/actions/workflows/ci.yml)     An backport of `django.tasks` - Django's built-in [Tasks framework](https://docs.djangoproject.com/en/stable/topics/tasks/). ## Installation ``` python -m pip install django-tasks ``` The first step is to add `django_tasks` to your `INSTALLED_APPS`. ```python INSTALLED_APPS = [ # ... "django_tasks", ] ``` Secondly, you'll need to configure a backend. This connects the tasks to whatever is going to execute them. If omitted, the following configuration is used: ```python TASKS = { "default": { "BACKEND": "django_tasks.backends.immediate.ImmediateBackend" } } ``` A few backends are included by default: - `django_tasks.backends.dummy.DummyBackend`: Don't execute the tasks, just store them. This is especially useful for testing. - `django_tasks.backends.immediate.ImmediateBackend`: Execute the task immediately in the current thread Prior to `0.12.0`, [`django-tasks-db`](https://github.com/RealOrangeOne/django-tasks-db) and [`django-tasks-rq`](https://github.com/RealOrangeOne/django-tasks-rq) were also included to provide database and RQ based backends. ## Usage ### Defining tasks A task is created with the `task` decorator. ```python from django_tasks import task @task() def calculate_meaning_of_life() -> int: return 42 ``` The task decorator accepts a few arguments to customize the task: - `priority`: The priority of the task (between -100 and 100. Larger numbers are higher priority. 0 by default) - `queue_name`: Whether to run the task on a specific queue - `backend`: Name of the backend for this task to use (as defined in `TASKS`) ```python modified_task = calculate_meaning_of_life.using(priority=10) ``` In addition to the above attributes, `run_after` can be passed to specify a specific time the task should run. #### Task context Sometimes the running task may need to know context about how it was enqueued. To receive the task context as an argument to your task function, pass `takes_context` to the decorator and ensure the task takes a `context` as the first argument. ```python from django_tasks import task, TaskContext @task(takes_context=True) def calculate_meaning_of_life(context: TaskContext) -> int: return 42 ``` The task context has the following attributes: - `task_result`: The running task result - `attempt`: The current attempt number for the task This API will be extended with additional features in future. ### Enqueueing tasks To execute a task, call the `enqueue` method on it: ```python result = calculate_meaning_of_life.enqueue() ``` The returned `TaskResult` can be interrogated to query the current state of the running task, as well as its return value. If the task takes arguments, these can be passed as-is to `enqueue`. ### Queue names By default, tasks are enqueued onto the "default" queue. When using multiple queues, it can be useful to constrain the allowed names, so tasks aren't missed. ```python TASKS = { "default": { "BACKEND": "django_tasks.backends.immediate.ImmediateBackend", "QUEUES": ["default", "special"] } } ``` Enqueueing tasks to an unknown queue name raises `InvalidTaskError`. To disable queue name validation, set `QUEUES` to `[]`. ### Retrieving task result When enqueueing a task, you get a `TaskResult`, however it may be useful to retrieve said result from somewhere else (another request, another task etc). This can be done with `get_result` (or `aget_result`): ```python result_id = result.id # Later, somewhere else... calculate_meaning_of_life.get_result(result_id) ``` A result `id` should be considered an opaque string, whose length could be up to 64 characters. ID generation is backend-specific. Only tasks of the same type can be retrieved this way. To retrieve the result of any task, you can call `get_result` on the backend: ```python from django_tasks import default_task_backend default_task_backend.get_result(result_id) ``` ### Return values If your task returns something, it can be retrieved from the `.return_value` attribute on a `TaskResult`. Accessing this property on an unsuccessful task (ie not `SUCCESSFUL`) will raise a `ValueError`. ```python assert result.status == TaskResultStatus.SUCCESSFUL assert result.return_value == 42 ``` If a result has been updated in the background, you can call `refresh` on it to update its values. Results obtained using `get_result` will always be up-to-date. ```python assert result.status == TaskResultStatus.READY result.refresh() assert result.status == TaskResultStatus.SUCCESSFUL