فریم‌ورک Tasks در Django: راهنمای کامل اجرای کارهای پس‌زمینه در Django 6.0

مقدمه

در بسیاری از وب‌اپلیکیشن‌ها، نیاز داریم کارهایی خارج از چرخهٔ request–response انجام دهیم؛ مثل ارسال ایمیل، پردازش فایل، تولید گزارش یا همگام‌سازی داده‌ها. Django 6.0 با معرفی Tasks Framework این نیاز را به‌صورت داخلی برطرف کرده است.

این فریم‌ورک مسئول:

• تعریف Task
• اعتبارسنجی
• صف‌بندی
• ذخیره و بازیابی نتیجه

اما اجرای Task را انجام نمی‌دهد. اجرای واقعی باید توسط یک Worker خارجی انجام شود.

مبانی کارهای پس‌زمینه

وقتی Django باید کاری را در پس‌زمینه انجام دهد، یک Task ایجاد کرده و آن را در Queue Store ذخیره می‌کند.

چرخهٔ اجرای Task

1. Task در صف قرار می‌گیرد
2. Worker آن را دریافت می‌کند
3. Task اجرا می‌شود
4. نتیجه در backend ذخیره می‌شود

پیکربندی Task Backend

Backend تعیین می‌کند:

• Taskها کجا ذخیره شوند
• Worker چگونه آن‌ها را دریافت کند
• نتایج چگونه ذخیره شوند

پیکربندی در تنظیمات:

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

ImmediateBackend

Backend پیش‌فرض است. Taskها را بلافاصله اجرا می‌کند، نه در پس‌زمینه.

مناسب برای:

• توسعهٔ اولیه
• محیط‌های بدون Worker
• تست‌ها

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

DummyBackend

Taskها را اجرا نمی‌کند؛ فقط نتیجه را ذخیره می‌کند و وضعیت را همیشه READY نگه می‌دارد.

مناسب برای:

• توسعه
• تست

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

مشاهدهٔ نتایج:

from django.tasks import default_task_backend

my_task.enqueue()
len(default_task_backend.results)

پاک‌سازی نتایج:

default_task_backend.clear()

Backendهای شخص ثالث

Backendهای داخلی Django فقط برای توسعه هستند. برای محیط production باید از backendهایی استفاده کنید که:

• صف پایدار (durable queue) دارند
• Worker اختصاصی دارند
• قابلیت retry و مدیریت خطا دارند

پیکربندی:

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

پشتیبانی Asynchronous

Backendها می‌توانند نسخهٔ async از متدهای خود را پیاده‌سازی کنند. نسخهٔ async با پیشوند a مشخص می‌شود (مثلاً enqueue() → aenqueue()).

دریافت Backendها

from django.tasks import task_backends

task_backends["default"]
task_backends["secondary"]

میان‌بر:

from django.tasks import default_task_backend

تعریف Task

Taskها با decorator @task روی توابع سطح ماژول تعریف می‌شوند.

مثال

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,
    )

خروجی decorator یک Task instance است.

تنظیم ویژگی‌های Task

@task(priority=2, queue_name="emails")
def email_users(...):
    ...

به‌صورت قراردادی، Taskها در فایل tasks.py قرار می‌گیرند.

Task Context

گاهی Task نیاز دارد بداند چگونه enqueue شده یا چندمین تلاش است. برای دریافت context باید takes_context=True را فعال کنید.

مثال

import logging
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 email. Task result id: {context.task_result.id}."
    )
    ...

جمع‌بندی

فریم‌ورک Tasks در Django 6.0 راهکاری قدرتمند و ساده برای اجرای کارهای پس‌زمینه ارائه می‌دهد. با پشتیبانی از backendهای مختلف، context، و نسخه‌های async، این سیستم پایه‌ای مناسب برای ساخت اپلیکیشن‌های مقیاس‌پذیر و سریع فراهم می‌کند. برای محیط production کافی است یک backend مناسب و Worker خارجی اضافه کنید تا Taskها به‌صورت واقعی اجرا شوند.

ویرایش Task قبل از enqueue

گاهی لازم است قبل از صف‌بندی یک Task، برخی ویژگی‌های آن را تغییر دهیم—مثلاً افزایش priority. Taskها immutable هستند و مستقیماً قابل تغییر نیستند. برای ایجاد نسخهٔ جدید از Task با تنظیمات متفاوت، از using() استفاده می‌شود.

مثال

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

نسخهٔ جدید Task ساخته می‌شود و نسخهٔ اصلی بدون تغییر باقی می‌ماند.

صف‌بندی Tasks (enqueue)

برای اضافه کردن Task به صف، از enqueue() استفاده می‌شود:

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

خروجی این متد یک TaskResult است که وضعیت و نتیجهٔ Task را نگه می‌دارد.

صف‌بندی در محیط async

نسخهٔ async این متد aenqueue() است:

result = await email_users.aenqueue(...)

محدودیت مهم: JSON Serialization

تمام آرگومان‌ها و خروجی Task باید قابل سریال‌سازی به JSON باشند. این یعنی:

• datetime → ❌
• tuple → تبدیل به list → ممکن است مشکل‌ساز شود
• model instance → ❌
• objectهای پیچیده → ❌

مثال خطا

>>> process_data.enqueue(datetime.now())
TypeError: Object of type datetime is not JSON serializable

مشکل round-trip JSON

مثلاً tuple تبدیل به list می‌شود و دیگر قابل استفاده به‌عنوان کلید دیکشنری نیست:

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

>>> result = double_dictionary.enqueue((1, 2, 3))
>>> result.status
FAILED

تراکنش‌ها و enqueue

Taskها معمولاً در فرآیندی جداگانه اجرا می‌شوند و از connection دیتابیس متفاوتی استفاده می‌کنند. اگر Task قبل از commit شدن تراکنش اجرا شود، ممکن است به داده‌ها دسترسی نداشته باشد.

مثال مشکل‌ساز

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

ممکن است Task قبل از commit اجرا شود و شیء را پیدا نکند.

راه‌حل: استفاده از transaction.on_commit()

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))

بازیابی نتایج Task

هر TaskResult یک id یکتا دارد. برای دریافت نتیجه در جای دیگر:

result = email_users.get_result(result_id)

یا از backend:

from django.tasks import default_task_backend

result = default_task_backend.get_result(result_id)

نسخهٔ async

result = await email_users.aget_result(result_id)

به‌روزرسانی وضعیت Task

TaskResult وضعیت را در لحظهٔ دریافت نشان می‌دهد. اگر Task بعداً تمام شود، باید آن را refresh کنید:

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

دریافت مقدار بازگشتی Task

اگر Task مقدار بازگرداند:

>>> result.return_value
42

اگر Task هنوز تمام نشده باشد:

ValueError: Task has not finished yet

مدیریت خطاها

اگر Task خطا دهد، اطلاعات خطا در result.errors ذخیره می‌شود.

مثال

>>> result.errors[0].exception_class

traceback نیز به‌صورت string ذخیره می‌شود:

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

جمع‌بندی

Django Tasks Framework امکانات قدرتمندی برای مدیریت کارهای پس‌زمینه ارائه می‌دهد—از ویرایش Task و صف‌بندی sync/async گرفته تا مدیریت تراکنش‌ها، بازیابی نتایج، و بررسی خطاها. با رعایت محدودیت‌های JSON و استفادهٔ صحیح از transaction.on_commit، می‌توانید سیستم‌های پس‌زمینهٔ قابل‌اعتماد و مقیاس‌پذیر بسازید.