사용 방법
- 설치 명령을 복사하고 자신의 에이전트 환경에 설치하기 전에 소스 패키지를 확인하세요.
- 필요한 도구, 인증 정보, 지원 환경은 원본 문서에서 확인하세요.
- 입력과 원하는 산출물을 설명하고 출처와 결과를 대조하세요.
npx skills add https://github.com/vintasoftware/django-ai-plugins --skill django-celery-expert시작 전 확인
- 설치에는 모델 구독, API 한도, 파일 접근 권한이 포함되지 않습니다.
- 별점 수와 설치 수는 활동 지표이며 독립적인 품질·보안 검증이 아닙니다.
원본 Skill 문서
원본 저장소의 제목, 예시, 코드, 표, 링크, 이미지를 유지해 표시합니다.
Django Celery Expert
Instructions
Step 1: Classify the Request
Identify the task category from the request:
- Django integration — transaction safety, ORM patterns, testing, request correlation → read
references/django-integration.md - Task design — new tasks, calling patterns, chains/groups/chords, idempotency → read
references/task-design-patterns.md - Configuration — broker setup, result backend, worker settings, queue routing → read
references/configuration-guide.md - Error handling — retries, backoff, dead letter queues, timeouts → read
references/error-handling.md - Periodic tasks — Celery Beat, crontab schedules, dynamic schedules, timezone handling → read
references/periodic-tasks.md - Monitoring — Flower, Prometheus, logging, debugging stuck tasks → read
references/monitoring-observability.md - Production deployment — scaling, supervision, containers, health checks → read
references/production-deployment.md
If the request spans multiple categories, read all relevant reference files before continuing.
Step 2: Read the Reference File(s)
Read each reference file identified in Step 1. Do not proceed to implementation without reading the relevant reference.
Step 3: Implement
Apply the patterns from the reference file. Before presenting the solution, verify:
- Task arguments are serializable (pass IDs, not model instances)
- Tasks with retries enabled are idempotent
- Errors are logged with context
- Long-running tasks have timeouts configured
Examples
Basic Background Task
Request: "Send welcome emails in the background after user registration"
python
# tasks.py
from celery import shared_task
from django.core.mail import send_mail
@shared_task(bind=True, max_retries=3)
def send_welcome_email(self, user_id):
from users.models import User
try:
user = User.objects.get(id=user_id)
send_mail(
subject="Welcome!",
message=f"Hi {user.name}, welcome to our platform!",
from_email="noreply@example.com",
recipient_list=[user.email],
)
except User.DoesNotExist:
pass
except Exception as exc:
raise self.retry(exc=exc, countdown=60 * (2 ** self.request.retries))
# views.py — queue only after the transaction commits
from django.db import transaction
def register(request):
user = User.objects.create(...)
transaction.on_commit(lambda: send_welcome_email.delay(user.id))
return redirect("dashboard")Task with Progress Tracking
Request: "Process a large CSV import with progress updates"
python
@shared_task(bind=True)
def import_csv(self, file_path, total_rows):
from myapp.models import Record
with open(file_path) as f:
reader = csv.DictReader(f)
for i, row in enumerate(reader):
Record.objects.create(**row)
if i % 100 == 0:
self.update_state(
state="PROGRESS",
meta={"current": i, "total": total_rows},
)
return {"status": "complete", "processed": total_rows}
# Poll progress
result = import_csv.AsyncResult(task_id)
if result.state == "PROGRESS":
progress = result.info.get("current", 0) / result.info.get("total", 1)Workflow with Chains
Request: "Process an order: validate inventory, charge payment, then send confirmation"
python
from celery import chain
@shared_task
def validate_inventory(order_id):
order = Order.objects.get(id=order_id)
if not order.items_in_stock():
raise ValueError("Items out of stock")
return order_id
@shared_task
def charge_payment(order_id):
order = Order.objects.get(id=order_id)
order.charge()
return order_id
@shared_task
def send_confirmation(order_id):
Order.objects.get(id=order_id).send_confirmation_email()
def process_order(order_id):
chain(
validate_inventory.s(order_id),
charge_payment.s(),
send_confirmation.s(),
).delay()
