django-generic-tasks
====================
Active Job for Django
Example usage
-------------
Define tasks in a ``tasks.py`` module within a Django app:
`my_app/tasks.py <django_generic_tasks_example/my_app/tasks.py>`_
::
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 <https://docs.djangoproject.com/en/4.0/topics/signals/#connecting-receiver-functions-1>`_, 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 <django_generic_tasks_example/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 <django_generic_tasks_example/django_generic_tasks_example/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 <https://django-ninja.rest-framework.com/>`_ 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.
.. image:: https://user-images.githubusercontent.com/11734309/165979039-df4d2bfe-4c38-4798-af2e-fd0792303608.png
:alt: autogenerated OpenAPI docs
Supported backends
------------------
ThreadingBackend
~~~~~~~~~~~~~~~~
Runs tasks in a new ``threading.Thread``.
CloudTasksBackend
~~~~~~~~~~~~~~~~~
Runs tasks using `Cloud Tasks <https://cloud.google.com/tasks>`_ `HTTP Target tasks <https://cloud.google.com/tasks/docs/creating-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