Tpl Backend Django Drf Celery

PROJECT: Django REST API + Celery

Skill gerado a partir do pack templates-claude-code. Arquivo de origem: backend/03-django-drf-celery.md. Use como baseline e adapte ao projeto antes de mudancas grandes.

Conteudo do template

STACK

PROJECT STRUCTURE

config/
├── settings/
│   ├── base.py           # Shared settings
│   ├── development.py    # Dev overrides
│   ├── production.py     # Prod overrides (env-driven)
│   └── test.py           # Test overrides (SQLite or test PG)
├── urls.py               # Root URL conf
└── celery.py             # Celery app instance
apps/
└── users/
    ├── migrations/       # Auto-generated
    ├── models.py         # Django ORM models
    ├── serializers.py    # DRF serializers
    ├── views.py          # ViewSets
    ├── urls.py           # Router registration
    ├── permissions.py    # Custom DRF permissions
    ├── tasks.py          # Celery tasks
    ├── services.py       # Business logic
    ├── filters.py        # django-filter FilterSets
    └── tests/
        ├── test_views.py
        ├── test_serializers.py
        ├── test_tasks.py
        └── factories.py  # factory_boy factories
manage.py
requirements/
├── base.txt
├── dev.txt
└── prod.txt

ARCHITECTURE RULES

• ViewSet → Serializer → Service → Model — ViewSets do not query the DB directly
• Business logic MUST live in services.py — not in ViewSets or serializers
• Serializer validate_* methods only validate format/consistency — no DB queries in validate
• DB queries belong in services.py or manager methods — not scattered in views
• Celery tasks are thin wrappers: receive IDs, call services, handle retry logic
• Custom manage.py commands import and call service functions directly
• Every model has a __str__ and class Meta with verbose_name and ordering

VIEWSET PATTERNS

# apps/users/views.py
from rest_framework import viewsets, status
from rest_framework.decorators import action
from rest_framework.response import Response
from rest_framework.permissions import IsAuthenticated, IsAdminUser
from .serializers import UserSerializer, UserCreateSerializer, PasswordChangeSerializer
from .services import UserService

class UserViewSet(viewsets.ModelViewSet):
    serializer_class = UserSerializer
    permission_classes = [IsAuthenticated]

    def get_queryset(self):
        if self.request.user.is_staff:
            return UserService.list_all()
        return UserService.list_active()

    def get_permissions(self):
        if self.action in ['destroy', 'list']:
            return [IsAdminUser()]
        return super().get_permissions()

    def get_serializer_class(self):
        if self.action == 'create':
            return UserCreateSerializer
        return self.serializer_class

    def perform_create(self, serializer):
        user = UserService.create_user(**serializer.validated_data)
        serializer.instance = user  # update serializer with saved instance

    @action(detail=False, methods=['get', 'patch'], url_path='me')
    def me(self, request):
        if request.method == 'GET':
            return Response(UserSerializer(request.user).data)
        serializer = UserSerializer(request.user, data=request.data, partial=True)
        serializer.is_valid(raise_exception=True)
        UserService.update_user(request.user, **serializer.validated_data)
        return Response(serializer.data)

    @action(detail=False, methods=['post'], url_path='change-password')
    def change_password(self, request):
        serializer = PasswordChangeSerializer(data=request.data)
        serializer.is_valid(raise_exception=True)
        UserService.change_password(request.user, **serializer.validated_data)
        return Response(status=status.HTTP_204_NO_CONTENT)

Serializer Validation

# apps/users/serializers.py
from rest_framework import serializers
from django.contrib.auth.password_validation import validate_password
from .models import User

class UserCreateSerializer(serializers.ModelSerializer):
    password = serializers.CharField(write_only=True, validators=[validate_password])
    password_confirm = serializers.CharField(write_only=True)

    class Meta:
        model = User
        fields = ['email', 'name', 'password', 'password_confirm']

    def validate(self, data):
        if data['password'] != data.pop('password_confirm'):
            raise serializers.ValidationError({'password_confirm': 'Passwords do not match.'})
        return data

Custom Permissions

# apps/users/permissions.py
from rest_framework.permissions import BasePermission

class IsOwnerOrAdmin(BasePermission):
    def has_object_permission(self, request, view, obj):
        return obj == request.user or request.user.is_staff

CELERY TASKS RULES

Task Definition Pattern

Always use bind=True for access to self (retry + task ID). Always define max_retries, default_retry_delay. Tasks must be idempotent — running twice should have the same outcome as running once.

# apps/users/tasks.py
from celery import shared_task
from celery.utils.log import get_task_logger
from .services import UserService

logger = get_task_logger(__name__)

@shared_task(
    bind=True,
    max_retries=3,
    default_retry_delay=60,       # seconds
    serializer='json',            # NEVER use pickle
    acks_late=True,               # ack only after task completes
    reject_on_worker_lost=True,   # re-queue if worker dies mid-task
    name='users.send_welcome_email',
)
def send_welcome_email(self, user_id: int) -> str:
    """Send welcome email to new user. Idempotent: safe to retry."""
    try:
        result = UserService.send_welcome_email(user_id=user_id)
        logger.info("Welcome email sent", extra={"user_id": user_id})
        return result
    except UserService.EmailAlreadySentError:
        logger.warning("Email already sent (idempotency check)", extra={"user_id": user_id})
        return "already_sent"
    except Exception as exc:
        logger.error("Failed to send email", exc_info=True, extra={"user_id": user_id})
        raise self.retry(exc=exc)

Celery Rules:

1. Always serializer='json' — never pickle (security risk)
2. acks_late=True on tasks that touch external services
3. Idempotency first — check if work was already done before doing it
4. IDs only — pass user_id: int, not Django model instances (not serializable cleanly)
5. Log with extra={} — structured logging, never f-string sensitive data
6. Retry with exponential backoff for external services: self.retry(exc=exc, countdown=2 ** self.request.retries * 60)

ROUTING TABLE

QUALITY GATES

Before opening a PR, verify ALL of the following:

• ruff check . && black --check . passes with zero issues
• pytest --tb=short -x passes with zero failures
• Coverage ≥ 80% for new service + view code
• python manage.py migrate --check — no unapplied migrations
• All new Celery tasks have a test with CELERY_TASK_ALWAYS_EAGER = True
• No raw QuerySet usage inside ViewSets — goes through services
• serializer.is_valid(raise_exception=True) used (not manual 400 handling)
• select_related / prefetch_related used where N+1 queries are possible
• Sensitive fields (password, token) marked write_only=True in serializers
• All Celery tasks registered in CELERY_TASK_ROUTES (dedicated queues in prod)

FORBIDDEN

• NEVER use pickle as Celery serializer — always json
• NEVER pass Django model instances to Celery tasks — pass IDs only
• NEVER query the database inside a Celery task's apply_async call site — batch IDs
• NEVER use bare except Exception: pass in tasks — always log or re-raise
• NEVER define business logic in serializer create() or update() — use services
• NEVER use User.objects.filter(...) directly in ViewSets — go through services
• NEVER run Celery with CELERY_ALWAYS_EAGER=True in production
• NEVER commit settings with DEBUG=True or hardcoded secrets
• NEVER skip migrations — every model change gets a migration
• NEVER use Django signals for critical business logic (hard to debug, order not guaranteed)