django-generic-tasks 0.5.1
Active Job for Django
Requires Python: >=3.9,<4.0

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