django-generic-tasks

Active Job for Django

Example usage

Define tasks in a tasks.py module within a Django app:

from django.core.mail import send_mail
from pydantic import BaseModel

import django_generic_tasks as tasks


# define task params as a Pydantic BaseModel
class EmailNotificationParams(BaseModel):
    subject: str
    content: str
    recipients: list[str]


# subclass Task and specify params type as a generic type argument and implement the run method
class EmailNotificationTask(tasks.Task[EmailNotificationParams]):
    def run(self):
        send_mail(
            subject=self.params.subject,
            message=self.params.content,
            from_email=None,
            recipient_list=self.params.recipients,
        )


if __name__ == "__main__":
    params = EmailNotificationParams(
        subject="Hello",
        content="Have a nice day",
        recipients=["alice@example.com", "bob@example.com"],
    )
    task = EmailNotificationTask(params)

    # run a task synchronously
    task.run()

    # run a task asynchronously using settings.TASKS_BACKEND
    task.start()

Registering tasks

Similar to signals, tasks have to be implicitly registered by ensuring they are imported during application startup. This can be done in the ready method in an application's AppConfig.

my_app/apps.py

from django.apps import AppConfig


class MyAppConfig(AppConfig):
    default_auto_field = "django.db.models.BigAutoField"
    name = "my_app"

    def ready(self):
        from . import tasks  # noqa: F401

HTTP endpoints

urls.py

from django.contrib import admin
from django.urls import include, path

urlpatterns = [
    path("admin/", admin.site.urls),
    path("tasks/", include("django_generic_tasks.urls")),
]

django-generic-tasks uses django-ninja to automatically expose API endpoints for running tasks. Each defined task gets its own API endpoint and uses the specified Pydantic BaseModel for parameter verification.

Supported backends

ThreadingBackend

Runs tasks in a new threading.Thread.

CloudTasksBackend

Runs tasks using Cloud Tasks HTTP Target tasks.

Configuration

`TASKS_API_AUTHENTICATION`

Specifies the Python path to a function or class whose instances are callable to use to authenticate incoming task API calls. The function or callable class should accept a single parameter request representing the incoming request.

Type: Optional[str]

Default: django_generic_tasks.security.BasicAuth

Examples:

my_app/authentication.py

# function
def authenticate(request):
    """Only allow requests from localhost"""
    return request.META["REMOTE_ADDR"] == "127.0.0.1"

# callable class instance
class Authenticator:
    def __call__(self, request):
        """Only allow requests from localhost"""
        return request.META["REMOTE_ADDR"] == "127.0.0.1"

my_app/settings.py

TASKS_API_AUTHENTICATION = "my_app.authentication.authenticate"
# or
TASKS_API_AUTHENTICATION = "my_app.authentication.Authenticator"

Built-in authentication methods:

django_generic_tasks.security.GoogleOIDCAuth - Enforces that incoming requests contain a Google-issued OIDC token in the authorization header. This can be automatically added to requests from Cloud Tasks and Cloud Scheduler.
django_generic_tasks.security.BasicAuth - Authenticates basic auth credentials using the Django authentication system
django_generic_tasks.security.NoAuth - No authentication, useful for development.

`TASKS_BACKEND`

The default backend to use to run tasks asynchronously.

Type: any class which implements the django_generic_tasks.backends.Backend protocol

yi-jiayu/django-generic-tasks