Djangos ramverk för uppgifter

New in Django 6.0:

För äldre Django-versioner finns bakåtporteringen django-tasks tillgänglig.

För en webbapplikation handlar det ofta om mer än att omvandla HTTP-begäranden till HTTP-svar. För vissa funktioner kan det vara fördelaktigt att köra kod utanför cykeln begäran–svar.

Där kommer bakgrundsuppgifter in.

Bakgrundsuppgifter kan lägga ut arbete som ska köras utanför cykeln begäran–svar, någon annanstans och eventuellt vid en senare tidpunkt. Det håller begäranden snabba, minskar latensen och förbättrar användarupplevelsen. En användare ska till exempel inte behöva vänta på att ett e-postmeddelande skickas innan sidan har lästs in.

Djangos ramverk för uppgifter gör det enkelt att definiera och köa sådant arbete. Det tillhandahåller ingen arbetsmekanism för att köra uppgifter. Den faktiska körningen måste hanteras av infrastruktur utanför Django, till exempel en separat process eller tjänst. Därför bör en uppgiftsbackend som kan köra uppgifter i den tjänsten utvärderas och konfigureras.

Grunderna för bakgrundsuppgifter

När arbete behöver utföras i bakgrunden skapar Django en Task, som lagras i Queue Store. Denna Task innehåller all metadata som behövs för att köra den samt en unik identifierare för att Django senare ska kunna hämta resultatet.

En Worker söker efter nya Tasks att köra i Queue Store. När en ny Task läggs till tar en Worker hand om den, kör den och sparar statusen och resultatet tillbaka i Queue Store. Dessa workers körs utanför livscykeln begäran–svar.

Konfigurera en uppgiftsbackend

Uppgiftsbackenden avgör hur och var uppgifter lagras för körning samt hur de körs. Olika uppgiftsbackendar har olika egenskaper och konfigurationsalternativ, vilket kan påverka applikationens prestanda och tillförlitlighet. Django innehåller inbyggda backendar, men dessa är endast avsedda för utveckling och testning.

Django hanterar definition, validering, köning och resultathantering av uppgifter, men inte körningen. Produktionsinstallationer behöver därför en backend eller worker-process som faktiskt kör köat arbete. Relevanta alternativ listas på sidan Community Ecosystem.

Uppgiftsbackendar konfigureras med inställningen TASKS i inställningsfilen. Även om de flesta applikationer bara behöver en backend stöds flera.

Omedelbar körning

Detta är standardbackenden om ingen annan anges i inställningsfilen. ImmediateBackend kör köade Tasks omedelbart i stället för i bakgrunden. Det gör det möjligt att successivt lägga till funktionalitet för bakgrundsuppgifter i en applikation innan den nödvändiga infrastrukturen finns.

Använd den genom att ange BACKEND till "django.tasks.backends.immediate.ImmediateBackend":

TASKS = {"default": {"BACKEND": "django.tasks.backends.immediate.ImmediateBackend"}}

ImmediateBackend kan även vara användbar i tester, för att slippa köra en verklig bakgrundsworker i testerna.

Dummy backend

DummyBackend kör inte alls köade Tasks utan lagrar i stället resultat för senare användning. Uppgiftsresultaten förblir för alltid i tillståndet READY.

Den här backenden är inte avsedd för användning i produktion – den tillhandahålls som en bekvämlighet som kan användas under utveckling och testning.

Använd den genom att ange BACKEND till "django.tasks.backends.dummy.DummyBackend":

TASKS = {"default": {"BACKEND": "django.tasks.backends.dummy.DummyBackend"}}

Resultaten för köade Tasks kan hämtas från backendens attribut results:

>>> from django.tasks import default_task_backend
>>> my_task.enqueue()
>>> len(default_task_backend.results)
1

Lagrade resultat kan rensas med metoden clear():

>>> default_task_backend.clear()
>>> len(default_task_backend.results)
0

Tredjepartsbackendar

Som nämndes i början av detta avsnitt innehåller Django backendar som endast lämpar sig för utveckling och testning. Produktionssystem bör förlita sig på backendar som tillhandahåller en worker-process och en beständig köimplementering. Tillgängliga tredjepartsbackendar listas på sidan Community Ecosystem och i Django Packages rutnät för ramverk för uppgifter.

Använd en extern uppgiftsbackend med Django genom att ange Pythons importsökväg som BACKEND i inställningen TASKS, så här:

TASKS = {
    "default": {
        "BACKEND": "path.to.backend",
    }
}

En uppgiftsbackend är en klass som ärver BaseTaskBackend. Den måste minst implementera BaseTaskBackend.enqueue(). När du bygger en egen backend kan du använda de inbyggda uppgiftsbackendarna som referensimplementationer. Koden finns i katalogen django/tasks/backends/ i Djangos källkod.

Asynkront stöd

Django har stöd under utveckling för asynkrona uppgiftsbackendar.

BaseTaskBackend har asynkrona varianter av alla grundmetoder. Enligt konvention har de asynkrona versionerna av alla metoder prefixet a. Argumenten är desamma för båda varianterna.

Hämta backendar

Backendar kan hämtas med anslutningshanteraren task_backends:

from django.tasks import task_backends

task_backends["default"]  # The default backend
task_backends["reserve"]  # Another backend

Standardbackenden finns tillgänglig som default_task_backend:

from django.tasks import default_task_backend

Definiera uppgifter

Tasks definieras med dekoratorn django.tasks.task() på en funktion på modulnivå:

from django.core.mail import send_mail
from django.tasks import task


@task
def email_users(emails, subject, message):
    return send_mail(
        subject=subject, message=message, from_email=None, recipient_list=emails
    )

Dekoratorns returvärde är en instans av Task.

Attribut för Task kan anpassas via dekoratorargumenten för @task:

from django.core.mail import send_mail
from django.tasks import task


@task(priority=2, queue_name="emails")
def email_users(emails, subject, message):
    return send_mail(
        subject=subject, message=message, from_email=None, recipient_list=emails
    )

Enligt konvention definieras Tasks i filen tasks.py, men detta upprätthålls inte.

Uppgiftskontext

Ibland behöver den körande Task känna till kontext om hur den köades och hur den körs. Detta kan nås genom att ta ett argument context, som är en instans av TaskContext.

För att ta emot uppgiftskontexten som argument till uppgiftsfunktionen skickar du takes_context när den definieras:

import logging
from django.core.mail import send_mail
from django.tasks import task

logger = logging.getLogger(__name__)


@task(takes_context=True)
def email_users(context, emails, subject, message):
    logger.debug(
        f"Attempt {context.attempt} to send user email. Task result id: {context.task_result.id}."
    )
    return send_mail(
        subject=subject, message=message, from_email=None, recipient_list=emails
    )

Ändra uppgifter

Innan Tasks köas kan det vara nödvändigt att ändra vissa parametrar för uppgiften, till exempel för att ge den högre prioritet än normalt.

En Task-instans kan inte ändras direkt. I stället kan en ändrad instans skapas med metoden using(), medan originalet lämnas oförändrat. Till exempel:

>>> email_users.priority
0
>>> email_users.using(priority=10).priority
10

Köa uppgifter

Lägg till uppgiften i kölagret så att den körs genom att anropa metoden enqueue() på den. Om uppgiften tar argument kan de skickas vidare som de är. Till exempel:

result = email_users.enqueue(
    emails=["[e-mail]"],
    subject="You have a message",
    message="Hello there!",
)

Detta returnerar en TaskResult, som kan användas för att hämta uppgiftens resultat när den har körts klart.

För att köa Tasks i ett async-sammanhang finns aenqueue() som en async-variant av enqueue().

Eftersom både uppgiftsargument och returvärden serialiseras till JSON måste de vara JSON-serialiserbara:

>>> process_data.enqueue(datetime.now())
Traceback (most recent call last):
...
TypeError: Object of type datetime is not JSON serializable

Argument måste också kunna gå tur och retur genom en cykel med json.dumps()/ json.loads() utan att byta typ. Överväg till exempel denna uppgift:

@task()
def double_dictionary(key):
    return {key: key * 2}

Med ImmediateBackend konfigurerad som standardbackend:

>>> result = double_dictionary.enqueue((1, 2, 3))
>>> result.status
FAILED
>>> result.errors[0].traceback
Traceback (most recent call last):
...
TypeError: unhashable type: 'list'

Uppgiften double_dictionary misslyckas eftersom tuppeln (1, 2, 3) efter JSON-omgången blir listan [1, 2, 3], som inte kan användas som en nyckel i en ordbok.

I allmänhet kan komplexa objekt som modellinstanser, eller inbyggda typer som datetime och tuple, inte användas i Tasks utan ytterligare konvertering.

Transaktioner

För de flesta backendar körs Tasks i en separat process med en annan databasanslutning. När en transaktion används kan workers, utan att vänta på att den bekräftas, börja bearbeta en uppgift som använder objekt som de ännu inte kan komma åt.

Överväg till exempel detta förenklade exempel:

@task
def my_task(thing_num):
    Thing.objects.get(num=thing_num)


with transaction.atomic():
    Thing.objects.create(num=1)
    my_task.enqueue(thing_num=1)

För att förhindra att my_task körs innan Thing har bekräftats till databasen använder du transaction.on_commit() och binder alla argument till enqueue() via functools.partial():

from functools import partial

from django.db import transaction

with transaction.atomic():
    Thing.objects.create(num=1)
    transaction.on_commit(partial(my_task.enqueue, thing_num=1))

Uppgiftsresultat

När en Task köas får du en TaskResult, men det är sannolikt användbart att hämta resultatet från någon annanstans (till exempel en annan begäran eller en annan Task).

Varje TaskResult har ett unikt id, som kan användas för att identifiera och hämta resultatet när koden som köade Tasken har avslutats.

Metoden get_result() kan hämta ett resultat baserat på dess id:

# Later, somewhere else...
result = email_users.get_result(result_id)

För att hämta ett TaskResult oavsett vilken sorts Task det kom från använder du metoden get_result() på backenden:

from django.tasks import default_task_backend

result = default_task_backend.get_result(result_id)

För att hämta resultat i ett async-sammanhang finns aget_result() som en async-variant av get_result() på både backenden och Task.

Vissa backendar, såsom den inbyggda ImmediateBackend, stöder inte get_result(). Att anropa get_result() på dessa backendar utlöser NotImplementedError.

Uppdatera resultat

En TaskResult innehåller statusen för en Tasks körning vid den tidpunkt då den hämtades. Om Tasken avslutas efter att get_result() har anropats uppdateras den inte.

Anropa metoden django.tasks.TaskResult.refresh() för att uppdatera värdena:

>>> result.status
RUNNING
>>> result.refresh()  # or await result.arefresh()
>>> result.status
SUCCESSFUL

Returvärden

Om uppgiftsfunktionen returnerar något kan det hämtas från attributet django.tasks.TaskResult.return_value:

>>> result.status
SUCCESSFUL
>>> result.return_value
42

Om Tasken inte har körts klart eller har misslyckats utlöses ValueError.

>>> result.status
RUNNING
>>> result.return_value
Traceback (most recent call last):
...
ValueError: Task has not finished yet

Fel

Om Tasken inte lyckas utan i stället utlöser ett undantag, antingen som en del av Tasken eller under körningen av den, sparas undantaget och stackspårningen i listan django.tasks.TaskResult.errors.

Varje post i errors är en TaskError som innehåller information om ett fel som utlöstes under körningen:

>>> result.errors[0].exception_class
<class 'ValueError'>

Observera att detta bara är undantagstypen och inte innehåller några andra värden. Stackspårningsinformationen reduceras till en sträng som du kan använda som hjälp vid felsökning:

>>> result.errors[0].traceback
Traceback (most recent call last):
...
TypeError: Object of type datetime is not JSON serializable