Standardmäßig ist der Prompt ausgewählt, der zuerst die Quelle prüft. Sie können zu einem direkten Befehl wechseln oder eine lokale Kopie herunterladen.
Quelldateien prüfen
Lesen Sie SKILL.md und alle von SkillsMP angezeigten Begleitdateien, bevor Sie sich für eine Installation entscheiden.
Mit Codex oder Claude installieren Kopieren Sie diesen Prompt, fügen Sie ihn in Codex, Claude oder einen anderen Assistant ein und lassen Sie die Skill-Seite prüfen und installieren.
Ein direkter Befehl überspringt den Prüf-Prompt. Prüfen Sie die Quelle, bevor Sie ihn ausführen.
# Start worker (development)
celery -A config worker --loglevel=info
# Start beat scheduler (periodic tasks)
celery -A config beat --loglevel=info --scheduler django_celery_beat.schedulers:DatabaseScheduler
# Combined worker + beat (dev only, never production)
celery -A config worker --beat --loglevel=info
# Production: multiple workers with concurrency
celery -A config worker --loglevel=warning --concurrency=4 -Q default,high_priority
Task Design Patterns
Basic Task
# apps/notifications/tasks.pyfrom celery import shared_task
import logging
logger = logging.getLogger(__name__)
@shared_task(name='notifications.send_welcome_email')defsend_welcome_email(user_id: int) -> None:
"""Send welcome email to newly registered user."""from apps.users.models import User
from apps.notifications.services import EmailService
try:
user = User.objects.get(pk=user_id)
except User.DoesNotExist:
logger.warning('send_welcome_email: user %s not found', user_id)
return# Idempotent — do not raise, task already impossible to complete
EmailService.send_welcome(user)
logger.info('Welcome email sent to user %s', user_id)
Retryable Task
@shared_task(
bind=True,
name='integrations.sync_to_crm',
max_retries=5,
default_retry_delay=60, # seconds before first retry
autoretry_for=(ConnectionError, TimeoutError),
retry_backoff=True, # exponential backoff
retry_backoff_max=600, # cap at 10 minutes
retry_jitter=True, # randomise to avoid thundering herd)defsync_contact_to_crm(self, contact_id: int) -> dict:
"""Sync contact to external CRM with retry on transient failures."""from apps.crm.services import CRMClient
try:
result = CRMClient().sync(contact_id)
return result
except CRMClient.RateLimitError as exc:
# Specific retry delay from response headerraiseself.retry(exc=exc, countdown=int(exc.retry_after))
Idempotent Task Pattern
Design tasks so they can safely run multiple times with the same inputs:
@shared_task(name='orders.mark_shipped')defmark_order_shipped(order_id: int, tracking_number: str) -> None:
"""Mark order as shipped — safe to run multiple times."""from apps.orders.models import Order
updated = Order.objects.filter(
pk=order_id,
status=Order.Status.PROCESSING, # Guard: only update if not already shipped
).update(
status=Order.Status.SHIPPED,
tracking_number=tracking_number,
)
ifnot updated:
logger.info('mark_order_shipped: order %s already shipped or not found', order_id)
Task with Soft Time Limit
from celery.exceptions import SoftTimeLimitExceeded
@shared_task(
bind=True,
name='reports.generate_pdf',
soft_time_limit=120,
time_limit=150,
)defgenerate_pdf_report(self, report_id: int) -> str:
"""Generate PDF report with graceful timeout handling."""from apps.reports.services import PDFGenerator
try:
path = PDFGenerator.build(report_id)
return path
except SoftTimeLimitExceeded:
# Clean up partial files before hard kill
PDFGenerator.cleanup(report_id)
raise
Calling Tasks
from datetime import timedelta
from django.utils import timezone
# Fire and forget (async)
send_welcome_email.delay(user.pk)
# Schedule in the future
send_reminder.apply_async(args=[user.pk], countdown=3600) # 1 hour from now
send_reminder.apply_async(args=[user.pk], eta=timezone.now() + timedelta(days=1))
# Apply with queue routing
sync_contact_to_crm.apply_async(args=[contact.pk], queue='high_priority')
# Run synchronously (tests / debugging only)
result = generate_pdf_report.apply(args=[report.pk])
from celery import chain, group, chord
# Chain: run tasks sequentially, passing results
pipeline = chain(
fetch_data.s(source_id),
transform_data.s(), # receives fetch_data result as first arg
load_to_warehouse.s(),
)
pipeline.delay()
# Group: run tasks in parallel
parallel = group(
send_welcome_email.s(user_id)
for user_id in new_user_ids
)
parallel.delay()
# Chord: parallel tasks + callback when all complete
result = chord(
group(process_chunk.s(chunk) for chunk in data_chunks),
aggregate_results.s(), # called with list of chunk results
)
result.delay()
Error Handling and Dead Letter Queue
# apps/core/tasks.pyfrom celery.signals import task_failure
@task_failure.connectdefon_task_failure(sender, task_id, exception, args, kwargs, traceback, einfo, **kw):
"""Log all task failures to Sentry / alerting."""import sentry_sdk
with sentry_sdk.new_scope() as scope:
scope.set_context('celery', {
'task': sender.name,
'task_id': task_id,
'args': args,
'kwargs': kwargs,
})
sentry_sdk.capture_exception(exception)
# Route failed tasks to dead-letter queue after max retries@shared_task(
bind=True,
max_retries=3,
name='payments.charge_card',
)defcharge_card(self, order_id: int) -> None:
from apps.payments.models import Order, FailedCharge
try:
_do_charge(order_id)
except Exception as exc:
ifself.request.retries >= self.max_retries:
# Persist to dead-letter table for manual review
FailedCharge.objects.create(
order_id=order_id,
error=str(exc),
task_id=self.request.id,
)
return# Don't raise — task is permanently failedraiseself.retry(exc=exc)
Testing Celery Tasks
Unit Testing (No Broker)
# tests/test_tasks.pyimport pytest
from unittest.mock import patch, MagicMock
from apps.notifications.tasks import send_welcome_email
classTestSendWelcomeEmail:
@pytest.mark.django_dbdeftest_sends_email_to_existing_user(self, user):
with patch('apps.notifications.services.EmailService') as mock_email:
send_welcome_email(user.pk)
mock_email.send_welcome.assert_called_once_with(user)
@pytest.mark.django_dbdeftest_skips_missing_user_gracefully(self):
"""Should not raise when user is deleted between enqueue and execute."""
send_welcome_email(99999) # Non-existent user — must not raise
Integration Testing with CELERY_TASK_ALWAYS_EAGER
# config/settings/test.py
CELERY_TASK_ALWAYS_EAGER = True# Run tasks synchronously in tests
CELERY_TASK_EAGER_PROPAGATES = True# Re-raise exceptions from tasks# tests/test_integration.py@pytest.mark.django_dbdeftest_registration_triggers_welcome_email(client):
with patch('apps.notifications.services.EmailService') as mock_email:
response = client.post('/api/users/', {
'email': 'new@example.com',
'password': 'strongpass123',
})
assert response.status_code == 201
mock_email.send_welcome.assert_called_once()
Testing Retries
@pytest.mark.django_dbdeftest_task_retries_on_connection_error():
with patch('apps.crm.services.CRMClient.sync') as mock_sync:
mock_sync.side_effect = ConnectionError('timeout')
with pytest.raises(ConnectionError):
sync_contact_to_crm.apply(args=[1], throw=True)
assert mock_sync.call_count == 1# First attempt only when eager
Monitoring
# Inspect active workers and queues
celery -A config inspect active
celery -A config inspect stats
celery -A config inspect reserved
# Check queue lengths (Redis)
redis-cli llen celery
# Flower: web-based real-time monitor
pip install flower
celery -A config flower --port=5555
Anti-Patterns
# BAD: Passing model instances — they may be stale by execution time
send_welcome_email.delay(user) # Never pass ORM objects
send_welcome_email.delay(user.pk) # Always pass PKs# BAD: Calling tasks synchronously in production views
result = generate_report.apply() # Blocks the request thread# BAD: Non-idempotent task without guards@shared_taskdefcharge_and_fulfill(order_id):
order.charge() # May charge twice if task retries!
order.fulfill()
# GOOD: Idempotent with status guard@shared_taskdefcharge_and_fulfill(order_id):
order = Order.objects.select_for_update().get(pk=order_id)
if order.status != Order.Status.PENDING:
return# Already processed
order.charge()
order.fulfill()
Production Checklist
Check
Setting
Worker restarts on crash
supervisord or systemd unit
CELERY_TASK_ACKS_LATE = True
Re-queue tasks on worker crash
CELERY_WORKER_PREFETCH_MULTIPLIER = 1
Fair distribution of long tasks
Separate queues per priority
-Q default,high_priority,low_priority
CELERY_TASK_SOFT_TIME_LIMIT set
Graceful timeout before hard kill
Sentry integration
Capture all task_failure signals
Flower or other monitor
Visibility into queue depths
Beat runs on single node only
Prevents duplicate scheduled task execution
Related Skills
django-patterns — ORM, service layer, and project structure
django-tdd — Testing Django models, views, and services
python-testing — pytest configuration and fixtures
