django-generic-tasks 0.3.2

Active Job for Django

django-generic-tasks

Active Job for Django

Example usage

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=["[e-mail]", "[e-mail]"],
    )
    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

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.

Example configuration

Create your own NinjaAPI instance and mount django_generic_tasks.api.router.

django_generic_tasks_example/api.py

from ninja import NinjaAPI

import django_generic_tasks.api

api = NinjaAPI()
api.add_router("/tasks/", django_generic_tasks.api.router)

Add your NinjaAPI instance in urls.py:

django_generic_tasks_example/urls.py

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

from .api import api

urlpatterns = [
    path("admin/", admin.site.urls),
    path("api/", api.urls),
]

Supported backends

ThreadingBackend

Runs tasks in a new threading.Thread.

CloudTasksBackend

Runs tasks using Cloud Tasks HTTP Target tasks.

Configuration

TASKS_API_AUTHENTICATION

Specifies what authentication should be required to run tasks via HTTP.

Type: Optional[str]

Supported values:

• "oidc" - Require API requests to contain a valid Google-issued OIDC token
• None (default) - No authentication

TASKS_BACKEND

The default backend to use to run tasks asynchronously.

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