from celery import Celery
from celery.schedules import crontab
app = Celery('myproject')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.conf.beat_schedule = {
'send-daily-report': {
'task': 'reports.tasks.send_daily_report',
'schedule': crontab(hour=8, minute=0), # Every day at 8 AM
},
'cleanup-old-sessions': {
'task': 'core.tasks.cleanup_sessions',
'schedule': crontab(hour=2, minute=0, day_of_week=1), # Monday 2 AM
},
'refresh-cache': {
'task': 'cache.tasks.refresh_popular_items',
'schedule': 300.0, # Every 5 minutes
},
}
from celery import shared_task
from django.core.mail import send_mail
from django.utils import timezone
from datetime import timedelta
@shared_task
def send_daily_report():
"""Generate and send daily analytics report."""
yesterday = timezone.now() - timedelta(days=1)
# Generate report data
from analytics.models import DailyStats
stats = DailyStats.objects.filter(date=yesterday.date()).first()
if stats:
send_mail(
subject=f'Daily Report - {yesterday.date()}',
message=f'Users: {stats.user_count}, Revenue: ${stats.revenue}',
from_email='reports@example.com',
recipient_list=['admin@example.com'],
)
Celery Beat schedules periodic tasks like cron jobs. I define schedules in settings or use the Django database scheduler. Tasks run at specified intervals or cron expressions. I use @periodic_task decorator or configure in CELERY_BEAT_SCHEDULE. For dynamic schedules, I use django-celery-beat with database-backed periodic tasks. I monitor task execution and failures. Beat must run as a separate process alongside workers. This enables features like daily reports, cache warming, or cleanup jobs without external cron.
