برای ساخت Command اختصاصی در Django کافی است داخل هر اپ پوشه management/commands بسازید، یک فایل پایتون با نام دلخواه اضافه کنید و کلاسی از BaseCommand پیاده سازی کنید؛ سپس می توانید آن را با python manage.py your_command اجرا و با cron، systemd timer یا Celery Beat زمان بندی کنید. در این راهنما همه مراحل با نمونه کد و نکات خطایابی توضیح داده شده است.

پیش نیاز و ساختار پوشه ها

فرض می کنیم پروژه جنگو شما فعال است و اپ مربوطه در INSTALLED_APPS ثبت شده. برای آماده کردن محل قرارگیری فرمان سفارشی:

  1. به ریشه اپ خود بروید (مثلا myapp).
  2. پوشه های management و داخل آن commands را بسازید.
  3. در هر دو پوشه یک فایل خالی __init__.py قرار دهید.
  4. فایل فرمان را داخل commands با نام دلخواه بسازید (مثلا cleanup_logs.py).
cd myproject/myapp
mkdir -p management/commands
touch management/__init__.py management/commands/__init__.py
touch management/commands/cleanup_logs.py

اسکلت اصلی یک management command

کد زیر حداقل ساختار لازم برای اجرای یک فرمان است. هدف: نمایش یک پیام ساده و نمونه استفاده از آرگومان.

# myapp/management/commands/hello.py
from django.core.management.base import BaseCommand, CommandError

class Command(BaseCommand):
    help = "نمونه ساده: چاپ یک پیام و عددی که کاربر وارد می کند"

    def add_arguments(self, parser):
        parser.add_argument("--times", type=int, default=1, help="تعداد تکرار پیام")

    def handle(self, *args, **options):
        times = options["times"]
        if times <= 0:
            raise CommandError("عدد times باید بزرگتر از صفر باشد.")
        for i in range(times):
            self.stdout.write(f"سلام از management command #{i+1}")
        self.stdout.write(self.style.SUCCESS("اجرا با موفقیت پایان یافت"))

اجرا:

python manage.py hello --times 3

نکات:

  • self.stdout.write برای چاپ خروجی استاندارد و self.style.SUCCESS برای رنگ و فرمت خط فرمان استفاده می شود.
  • CommandError باعث کد خروجی غیرصفر می شود که برای تشخیص خطا در زمان بندی مفید است.

یک مثال عملی: پاکسازی لاگ های قدیمی

در عمل فرمان ها برای کارهای نگهداری دیتابیس یا یکپارچه سازی استفاده می شوند. فرض کنید مدلی به نام AuditLog دارید که رکوردهای قدیمی آن باید حذف شوند.

# myapp/models.py
from django.db import models

class AuditLog(models.Model):
    action = models.CharField(max_length=100)
    created_at = models.DateTimeField(auto_now_add=True)

    def __str__(self):
        return f"{self.action} at {self.created_at}"

فرمان پاکسازی با آرگومان قابل تنظیم days:

# myapp/management/commands/cleanup_logs.py
import logging
from datetime import timedelta
from django.utils import timezone
from django.core.management.base import BaseCommand, CommandError
from django.db import transaction
from myapp.models import AuditLog

logger = logging.getLogger(__name__)

class Command(BaseCommand):
    help = "حذف لاگ هاي قديمي بر اساس تعداد روز"

    def add_arguments(self, parser):
        parser.add_argument("--days", type=int, default=90, help="حذف رکوردهاي قديمي تر از اين تعداد روز")
        parser.add_argument("--dry-run", action="store_true", help="فقط شمارش، بدون حذف")

    def handle(self, *args, **options):
        days = options["days"]
        if days <= 0:
            raise CommandError("days بايد بزرگتر از صفر باشد.")

        threshold = timezone.now() - timedelta(days=days)
        qs = AuditLog.objects.filter(created_at__lt=threshold)
        count = qs.count()

        self.stdout.write(f"{count} رکورد قديمي تر از {days} روز پيدا شد.")
        logger.info("cleanup_logs: %s records older than %s days found", count, days)

        if options["dry_run"]:
            self.stdout.write(self.style.WARNING("dry-run فعال است؛ رکوردي حذف نشد."))
            return

        # اتميك براي امنيت بيشتر
        with transaction.atomic():
            deleted, _ = qs.delete()

        self.stdout.write(self.style.SUCCESS(f"{deleted} رکورد حذف شد."))
        logger.info("cleanup_logs: %s records deleted", deleted)

روش بررسی نتیجه:

  • اجرا با گزارش خشک: python manage.py cleanup_logs –days 30 –dry-run
  • بررسی لاگ ها: با پیکربندی logging در settings.py پیام های logger را در فایل ثبت کنید.
  • بررسی کد خروجی: در لینوکس echo $? بعد از اجرا صفر یا غیرصفر بودن را نشان می دهد.

افزودن لاگ و تنظیم verbosity

بهتر است خروجی فرمان هم برای انسان در ترمینال خوانا باشد و هم در فایل لاگ ذخیره شود. پیشنهاد:

  • از logging.getLogger استفاده کنید تا تنظیمات سراسری لاگ به کار رود.
  • از self.stdout.write برای نمایش فوری خروجی استفاده کنید.
  • با سوئیچ های جنگو مثل –verbosity=2 در زمان اجرا، جزییات بیشتری چاپ کنید.
python manage.py cleanup_logs --days 60 --verbosity=2

زمان بندی اجرا با cron (لینوکس و مک)

cron رایج ترین روش اجرای زمان بندی شده است. از آنجا که محیط cron مینیمال است، بهترین کار استفاده از یک اسکریپت میانجی برای فعال کردن virtualenv و تنظیم مسیر پروژه است.

# myproject/run_cleanup.sh
#!/usr/bin/env bash
set -e
cd /path/to/myproject
source .venv/bin/activate
# در صورت نياز: export DJANGO_SETTINGS_MODULE=myproject.settings
python manage.py cleanup_logs --days 90 --verbosity=1

قابل اجرا کردن و افزودن به crontab:

chmod +x /path/to/myproject/run_cleanup.sh
crontab -e

# اجراي هر شب ساعت 02:30 و جلوگيري از همپوشاني با flock
30 2 * * * flock -n /tmp/cleanup_logs.lock /path/to/myproject/run_cleanup.sh >> /path/to/logs/cleanup.log 2>&1

توضیحات:

  • flock از اجرای همزمان چندباره جلوگیری می کند.
  • >> خروجی را به فایل لاگ اضافه می کند و 2>&1 خطا را هم به همانجا می فرستد.
  • اگر پروژه چند تنظیم دارد، از –settings یا متغیر DJANGO_SETTINGS_MODULE استفاده کنید.

زمان بندی با systemd timer (لینوکس)

systemd کنترل دقیق تر، لاگ یکپارچه و مدیریت وابستگی ها را فراهم می کند. دو فایل واحد (unit) بسازید:

# /etc/systemd/system/django-cleanup.service
[Unit]
Description=Django cleanup_logs command

[Service]
Type=oneshot
WorkingDirectory=/path/to/myproject
ExecStart=/path/to/myproject/.venv/bin/python manage.py cleanup_logs --days 90
User=www-data
Group=www-data

# /etc/systemd/system/django-cleanup.timer
[Unit]
Description=Run django cleanup_logs nightly

[Timer]
OnCalendar=*-*-* 02:30:00
Persistent=true

[Install]
WantedBy=timers.target

sudo systemctl daemon-reload
sudo systemctl enable --now django-cleanup.timer
sudo systemctl status django-cleanup.timer

خروجی و خطاها در journalctl قابل مشاهده است:

journalctl -u django-cleanup.service --since "1 day ago"

زمان بندی در ویندوز (Task Scheduler)

در ویندوز از Task Scheduler استفاده کنید:

  1. Create Basic Task را انتخاب کنید.
  2. Trigger را تنظیم کنید (مثلا روزانه).
  3. Action را روی Start a program بگذارید و مسیر python.exe داخل venv را انتخاب کنید.
  4. در قسمت Arguments بنویسید: manage.py cleanup_logs –days 90
  5. Start in را روی مسیر ریشه پروژه تنظیم کنید.

چه زمانی به Celery Beat نیاز داریم؟

اگر کار شما طولانی است، به صف بندی، تاخیر انعطاف پذیر، یا مقیاس افقی نیاز دارد، زمان بندی داخلی سیستم عامل کافی نیست. Celery به همراه Celery Beat زمان بندی مبتنی بر صف را فراهم می کند و شکست ها را بهتر مدیریت می کند. اما برای اسکریپت های مدیریتی سبک مثل پاکسازی، cron یا systemd ساده تر و کم هزینه تر است.

نمونه زمان بندی با Celery Beat (خلاصه)

در صورتی که از Celery استفاده می کنید، می توانید منطق فرمان را به یک تسک منتقل کنید و Beat را تنظیم کنید:

# myapp/tasks.py
from celery import shared_task
from django.utils import timezone
from datetime import timedelta
from myapp.models import AuditLog

@shared_task
def cleanup_logs_task(days=90):
    threshold = timezone.now() - timedelta(days=days)
    return AuditLog.objects.filter(created_at__lt=threshold).delete()[0]

# celery beat schedule (نمونه)
from celery.schedules import crontab
CELERY_BEAT_SCHEDULE = {
    "cleanup-logs-nightly": {
        "task": "myapp.tasks.cleanup_logs_task",
        "schedule": crontab(hour=2, minute=30),
        "args": (90,),
    }
}

نکته: راه اندازی Celery نیازمند بروکر پیام و اجرای worker است و از حوصله این راهنما خارج است.

نکات مهم برای پایداری و امنیت

  • ایدمپوتنت بودن: اجرای تکراری نباید اثر ناخواسته ایجاد کند. حذف با فیلتر زمان یا وضعیت ایمن است.
  • جلوگیری از همزمانی: در cron از flock استفاده کنید. در سطح کد می توانید قفل دیتابیس یا قفل مبتنی بر cache پیاده کنید.
  • ترنزاکشن: عملیات چندمرحله ای را داخل transaction.atomic اجرا کنید تا حالت بینابینی نماند.
  • مدیریت خطا: استثناهای قابل پیش بینی را به CommandError تبدیل کنید تا ابزار زمان بندی متوجه شکست شود.
  • تنظیم منابع: در کارهای حجیم از QuerySet.iterator یا پردازش دسته ای (batch) استفاده کنید تا حافظه مصرف نشود.
  • پیکربندی: مقادیر حساس یا پویای فرمان را از settings یا متغیرهای محیطی بخوانید، نه اینکه در کد ثابت کنید.

خطاهای رایج و روش رفع

  • ModuleNotFoundError برای اپ: بررسی کنید اپ در INSTALLED_APPS باشد و management/commands دارای __init__.py باشد.
  • timezone ناهمخوان: threshold را با django.utils.timezone.now بسازید، نه datetime.now.
  • اجرا نشدن در cron: مسیرهای مطلق، فعال سازی venv و سطح دسترسی فایل ها را بررسی کنید. دستور را به صورت دستی با همان کاربر cron تست کنید.
  • همپوشانی اجرا: از flock یا قفل نرم افزاری استفاده کنید.
  • اختلال بعد از مهاجرت دیتابیس: فرمان هایی که به مدل ها وابسته اند را پس از اعمال migrations اجرا کنید و نسخه بندی هماهنگ داشته باشید.

گام بعدی چیست؟

یک فرمان کوچک بسازید که وضعیت سلامت یک سرویس داخلی را بررسی و نتیجه را لاگ کند، سپس آن را با cron یا systemd زمان بندی کنید. پس از آن، فرمان های وابسته به دیتابیس را با ترنزاکشن، لاگ و آرگومان های قابل تنظیم تکمیل کنید تا برای محیط عملیاتی آماده باشند.