Python Celery Redis: الـ8 خطوات الرئيسية للانتقال من الصفر إلى التشغيل

pip install django-celery-beat

# settings.py
INSTALLED_APPS = [..., "django_celery_beat"]
CELERY_BEAT_SCHEDULER = "django_celery_beat.schedulers:DatabaseScheduler"

# تشغيل التهجيرات
python manage.py migrate

# يقرأ Beat الجدول الزمني من قاعدة البيانات كل 5 ثوانٍ
celery -A myproject beat -l info

from django_celery_beat.models import CrontabSchedule, PeriodicTask

# 1. تحديد التوقيت
schedule, _ = CrontabSchedule.objects.get_or_create(
    minute="30",
    hour="9",
    day_of_week="mon-fri",
    day_of_month="*",
    month_of_year="*",
    timezone="Europe/Paris"
)

# 2. إنشاء المهمة الدورية
PeriodicTask.objects.create(
    crontab=schedule,
    name="Newsletter Acme Corp",
    task="app.tasks.send_newsletter",
    args=json.dumps(["acme_corp"]),
    kwargs=json.dumps({"template": "weekly"}),
    enabled=True,
    one_off=False,
)

# يعرض Django admin بالفعل النماذج
# يمكنك الذهاب مباشرة إلى /admin/django_celery_beat/periodictask/
# للإضافة والتعديل والتعطيل عبر واجهة المستخدم

@app.post("/schedules")
def create_schedule(data: ScheduleCreate, db: Session = Depends(get_db)):
    schedule = CrontabSchedule.objects.create(
        minute=data.minute, hour=data.hour, ...
    )
    PeriodicTask.objects.create(
        name=data.name,
        crontab=schedule,
        task=data.task_name,
        args=json.dumps(data.args)
    )
    return {"status": "created"}

@app.delete("/schedules/{name}")
def delete_schedule(name: str):
    PeriodicTask.objects.filter(name=name).delete()

pip install celery-redbeat

# celery_app.py
celery_app.conf.beat_scheduler = "redbeat.RedBeatScheduler"
celery_app.conf.redbeat_redis_url = "redis://redis:6379/2"

# إنشاء جدول زمني
from redbeat import RedBeatSchedulerEntry
from celery.schedules import crontab

entry = RedBeatSchedulerEntry(
    name="my-task",
    task="app.tasks.my_task",
    schedule=crontab(hour=9, minute=0),
    app=celery_app,
    args=["arg1"]
)
entry.save()

# حذف
entry.delete()

# نموذج قاعدة البيانات
class Newsletter(models.Model):
    name = models.CharField(max_length=200)
    schedule_cron = models.CharField(max_length=100)        # "0 9 * * mon"
    enabled = models.BooleanField(default=True)
    
    def save(self, *args, **kwargs):
        super().save(*args, **kwargs)
        self.sync_celery_beat()
    
    def sync_celery_beat(self):
        m, h, dom, mon, dow = self.schedule_cron.split()
        schedule, _ = CrontabSchedule.objects.get_or_create(
            minute=m, hour=h, day_of_month=dom,
            month_of_year=mon, day_of_week=dow
        )
        PeriodicTask.objects.update_or_create(
            name=f"newsletter_{self.id}",
            defaults={
                "crontab": schedule,
                "task": "app.tasks.send_newsletter",
                "args": json.dumps([self.id]),
                "enabled": self.enabled
            }
        )

+---------------+
| Celery Beat   |  cron mon 9h
| (1 instance)  |
+-------+-------+
        |
        v
+---------------+      +-----------+
| dispatch_     |--->  | Redis     |
| weekly_emails |      | broker    |
| (1 task)      |      +-----+-----+
+---------------+            |
                             v
                +------------+------------+
                |                         |
        +-------v------+         +--------v-------+
        | Worker emails|         | Worker ML      |
        | (10 procs)   |         | (2 procs, GPU) |
        +-------+------+         +--------+-------+
                |                         |
                v                         v
        +-------+------+         +--------+-------+
        | SendGrid     |         | Modele Recommandation|
        | (rate limit) |         | + DB historique|
        +--------------+         +----------------+
                |
                v
        +-------+------+
        | Webhook      |
        | open/click   |
        +--------------+

recommender/
├── docker-compose.yml
├── Dockerfile
├── requirements.txt
├── app/
│   ├── celery_app.py
│   ├── config.py
│   ├── tasks/
│   │   ├── orchestrator.py      # dispatch_weekly_emails
│   │   ├── ml.py                # compute_recommendations
│   │   ├── email.py             # render + send
│   │   └── tracking.py          # webhooks
│   ├── models/                  # SQLAlchemy
│   ├── ml/
│   │   └── recommender.py
│   └── templates/
│       └── weekly.html
└── tests/

class User(Base):
    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str]
    active: Mapped[bool] = mapped_column(default=True)
    last_recos_sent: Mapped[datetime | None]

class Product(Base):
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]
    category: Mapped[str]
    embedding: Mapped[list[float]]    # متجه لتعلم الآلة

class Order(Base):
    id: Mapped[int] = mapped_column(primary_key=True)
    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
    product_id: Mapped[int] = mapped_column(ForeignKey("products.id"))
    created_at: Mapped[datetime]

class EmailJob(Base):
    """تتبع الإرسالات للامركزية"""
    id: Mapped[int] = mapped_column(primary_key=True)
    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
    week_key: Mapped[str]                  # "2026-W21"
    status: Mapped[str]                     # pending|sent|failed|opened|clicked
    sendgrid_msg_id: Mapped[str | None]
    created_at: Mapped[datetime] = mapped_column(default=datetime.utcnow)
    
    __table_args__ = (UniqueConstraint("user_id", "week_key"),)

celery_app.conf.task_queues = (
    Queue("orchestrator"),
    Queue("ml"),
    Queue("emails"),
    Queue("tracking"),
)

celery_app.conf.task_routes = {
    "app.tasks.orchestrator.*": {"queue": "orchestrator"},
    "app.tasks.ml.*": {"queue": "ml"},
    "app.tasks.email.*": {"queue": "emails"},
    "app.tasks.tracking.*": {"queue": "tracking"},
}

celery_app.conf.beat_schedule = {
    "weekly-recos": {
        "task": "app.tasks.orchestrator.dispatch_weekly_emails",
        "schedule": crontab(hour=9, minute=0, day_of_week="mon")
    }
}

# redis.conf
save 900 1          # لقطة إذا كان هناك تعديل واحد على الأقل في 15 دقيقة
save 300 10         # لقطة إذا كان هناك 10 تعديلات على الأقل في 5 دقائق
save 60 10000       # لقطة إذا كان هناك 10000 تعديل على الأقل في دقيقة واحدة

dbfilename dump.rdb
dir /var/lib/redis

# redis.conf
appendonly yes
appendfilename "appendonly.aof"

appendfsync always       # بطيء لكنه بدون فقدان
appendfsync everysec     # الافتراضي: فقدان 0-1 ثانية
appendfsync no           # يقرر نظام التشغيل

auto-aof-rewrite-percentage 100
auto-aof-rewrite-min-size 64mb

# موصى به في الإنتاج: تفعيل الاثنين
save 900 1
appendonly yes
appendfsync everysec

# عند إعادة التشغيل: يحمل Redis AOF (الأحدث)

# حد الذاكرة
maxmemory 2gb

# السياسة عند الامتلاء
maxmemory-policy allkeys-lru        # يحذف الأقل استخدامًا مؤخرًا (ذاكرة التخزين المؤقت)

# لقطة يدوية
redis-cli BGSAVE              # غير متزامن
redis-cli SAVE                # متزامن (يحظر!)

# نسخ التفريغ
cp /var/lib/redis/dump.rdb /backup/dump-$(date +%F).rdb

# استعادة
systemctl stop redis
cp /backup/dump-2026-05-27.rdb /var/lib/redis/dump.rdb
systemctl start redis

# على النسخة المتماثلة
replicaof master.redis.local 6379

# التحقق
redis-cli INFO replication
# role:slave
# master_link_status:up

# تجاوز الفشل التلقائي: Redis Sentinel
# التقسيم: Redis Cluster