Backend Orchestrator Skill

Purpose

Manages all backend development tasks including API design, service implementation, database operations, business logic, and backend testing using Django/Python by default.

Tier

Tier 0-2 - Pre-Launch to Growth

When to Use

• Building REST APIs
• Implementing business logic
• Database design and operations
• Service integrations
• Authentication/authorization
• Background job processing

Technology Stack

Default Stack

framework: "Django 4.2+"
language: "Python 3.11+"
database: "PostgreSQL"
cache: "Redis"
task_queue: "Celery"
api_framework: "Django REST Framework"
testing: "pytest + factory_boy"
authentication: "Django Auth + JWT"

Alternative Options

• FastAPI for microservices
• Flask for lightweight APIs
• Node.js/Express for JavaScript teams
• MySQL for different DB needs

API Architecture

RESTful Design

# apps/payments/views.py
from rest_framework import viewsets, status
from rest_framework.decorators import action
from rest_framework.response import Response

class PaymentViewSet(viewsets.ModelViewSet):
    """
    API endpoint for payment operations
    """
    queryset = Payment.objects.all()
    serializer_class = PaymentSerializer
    permission_classes = [IsAuthenticated]

    def create(self, request):
        """POST /api/payments/"""
        # 1. Validate input
        serializer = self.get_serializer(data=request.data)
        serializer.is_valid(raise_exception=True)

        # 2. Business logic
        payment = self.perform_payment(serializer.validated_data)

        # 3. Response
        return Response(
            PaymentSerializer(payment).data,
            status=status.HTTP_201_CREATED
        )

    @action(detail=True, methods=['post'])
    def refund(self, request, pk=None):
        """POST /api/payments/{id}/refund/"""
        payment = self.get_object()
        refund = payment.process_refund(request.data.get('amount'))
        return Response(RefundSerializer(refund).data)

API Versioning

# urls.py
urlpatterns = [
    path('api/v1/', include('apps.v1.urls')),
    path('api/v2/', include('apps.v2.urls')),
]

Database Design

Model Structure

# apps/payments/models.py
from django.db import models
from django.contrib.auth import get_user_model

class Payment(models.Model):
    """Payment transaction model"""

    STATUS_CHOICES = [
        ('pending', 'Pending'),
        ('processing', 'Processing'),
        ('completed', 'Completed'),
        ('failed', 'Failed'),
        ('refunded', 'Refunded'),
    ]

    # Relations
    user = models.ForeignKey(
        get_user_model(),
        on_delete=models.PROTECT,
        related_name='payments'
    )

    # Fields
    amount = models.DecimalField(
        max_digits=10,
        decimal_places=2,
        validators=[MinValueValidator(0.01)]
    )
    currency = models.CharField(max_length=3, default='USD')
    status = models.CharField(
        max_length=20,
        choices=STATUS_CHOICES,
        default='pending'
    )

    # Metadata
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)

    class Meta:
        db_table = 'payments'
        indexes = [
            models.Index(fields=['user', 'status']),
            models.Index(fields=['created_at']),
        ]

    def process(self):
        """Process the payment"""
        # Business logic here
        pass

Migration Strategy

# apps/payments/migrations/0001_initial.py
class Migration(migrations.Migration):
    dependencies = []

    operations = [
        migrations.CreateModel(
            name='Payment',
            fields=[
                # Field definitions
            ],
        ),
        migrations.RunSQL(
            "CREATE INDEX idx_payment_status ON payments(status);",
            reverse_sql="DROP INDEX idx_payment_status;"
        ),
    ]

Service Layer

Business Logic

# apps/payments/services.py
from typing import Optional
from decimal import Decimal

class PaymentService:
    """Encapsulates payment business logic"""

    def create_payment(
        self,
        user: User,
        amount: Decimal,
        payment_method: str
    ) -> Payment:
        """
        Create and process a payment
        """
        # 1. Validate
        self._validate_payment_method(payment_method)
        self._check_user_limits(user, amount)

        # 2. Create payment
        payment = Payment.objects.create(
            user=user,
            amount=amount,
            status='pending'
        )

        # 3. Process asynchronously
        process_payment_task.delay(payment.id)

        return payment

    def _validate_payment_method(self, method: str) -> None:
        """Validate payment method"""
        if method not in ALLOWED_METHODS:
            raise ValidationError(f"Invalid payment method: {method}")

    def _check_user_limits(self, user: User, amount: Decimal) -> None:
        """Check user transaction limits"""
        daily_total = self._get_daily_total(user)
        if daily_total + amount > user.daily_limit:
            raise ValidationError("Daily limit exceeded")

Authentication & Authorization

JWT Authentication

# apps/authentication/views.py
from rest_framework_simplejwt.views import TokenObtainPairView
from rest_framework_simplejwt.serializers import TokenObtainPairSerializer

class CustomTokenSerializer(TokenObtainPairSerializer):
    @classmethod
    def get_token(cls, user):
        token = super().get_token(user)
        token['email'] = user.email
        token['role'] = user.role
        return token

class LoginView(TokenObtainPairView):
    serializer_class = CustomTokenSerializer

Permission Classes

# apps/core/permissions.py
from rest_framework import permissions

class IsOwnerOrReadOnly(permissions.BasePermission):
    """
    Custom permission to only allow owners of an object to edit it.
    """
    def has_object_permission(self, request, view, obj):
        # Read permissions for any request
        if request.method in permissions.SAFE_METHODS:
            return True

        # Write permissions only for owner
        return obj.user == request.user

Testing Strategy

All 8 Test Types

# tests/test_payment_api.py
import pytest
from rest_framework.test import APIClient

class TestPaymentAPI:
    """Payment API test suite"""

    # 1. Valid: Happy path
    def test_create_payment_success(self, api_client, user):
        response = api_client.post('/api/payments/', {
            'amount': 100.00,
            'method': 'credit_card'
        })
        assert response.status_code == 201

    # 2. Error: Handle failures
    def test_payment_gateway_error(self, api_client, mock_gateway_error):
        response = api_client.post('/api/payments/', {...})
        assert response.status_code == 502

    # 3. Invalid: Input validation
    def test_negative_amount_rejected(self, api_client):
        response = api_client.post('/api/payments/', {
            'amount': -10.00
        })
        assert response.status_code == 400

    # 4. Edge: Boundary conditions
    def test_minimum_payment_amount(self, api_client):
        response = api_client.post('/api/payments/', {
            'amount': 0.01
        })
        assert response.status_code == 201

    # 5. Functional: Business logic
    def test_daily_limit_enforcement(self, api_client, user_at_limit):
        response = api_client.post('/api/payments/', {...})
        assert response.status_code == 400

    # 6. Visual: API response format
    def test_response_structure(self, api_client):
        response = api_client.get('/api/payments/')
        assert 'results' in response.data
        assert 'count' in response.data

    # 7. Performance: Response time
    def test_api_response_time(self, api_client, django_benchmark):
        with django_benchmark:
            response = api_client.get('/api/payments/')
        assert django_benchmark.elapsed < 0.2

    # 8. Security: Authorization
    def test_unauthorized_access_denied(self, api_client):
        api_client.logout()
        response = api_client.get('/api/payments/')
        assert response.status_code == 401

Background Tasks

Celery Tasks

# apps/payments/tasks.py
from celery import shared_task
from django.core.mail import send_mail

@shared_task(bind=True, max_retries=3)
def process_payment_task(self, payment_id):
    """Process payment asynchronously"""
    try:
        payment = Payment.objects.get(id=payment_id)
        payment.process()
        send_payment_confirmation.delay(payment_id)
    except Payment.DoesNotExist:
        raise
    except Exception as exc:
        # Retry with exponential backoff
        raise self.retry(exc=exc, countdown=2 ** self.request.retries)

@shared_task
def send_payment_confirmation(payment_id):
    """Send confirmation email"""
    payment = Payment.objects.get(id=payment_id)
    send_mail(
        'Payment Confirmation',
        f'Your payment of {payment.amount} has been processed.',
        'noreply@example.com',
        [payment.user.email],
    )

Error Handling

Global Exception Handler

# apps/core/exceptions.py
from rest_framework.views import exception_handler
from rest_framework.response import Response

def custom_exception_handler(exc, context):
    response = exception_handler(exc, context)

    if response is not None:
        custom_response_data = {
            'error': {
                'status': response.status_code,
                'message': response.data.get('detail', str(exc)),
                'timestamp': datetime.now().isoformat(),
                'path': context['request'].path
            }
        }
        response.data = custom_response_data

    return response

Performance Optimization

Database Optimization

# Query optimization
payments = Payment.objects.select_related('user') \
                         .prefetch_related('items') \
                         .filter(status='completed') \
                         .only('id', 'amount', 'created_at')

# Database indexing
class Meta:
    indexes = [
        models.Index(fields=['user', 'status']),
        models.Index(fields=['created_at']),
    ]

Caching Strategy

from django.core.cache import cache

def get_user_payment_summary(user_id):
    cache_key = f'payment_summary_{user_id}'
    summary = cache.get(cache_key)

    if summary is None:
        summary = calculate_payment_summary(user_id)
        cache.set(cache_key, summary, timeout=3600)

    return summary

Quality Standards

• API response time: < 200ms (p95)
• Test coverage: > 80%
• Code complexity: < 10 (McCabe)
• Database queries: < 10 per request
• Error rate: < 0.1%
• Security: OWASP Top 10 compliance

File Structure

apps/
├── authentication/
├── payments/
│   ├── __init__.py
│   ├── models.py
│   ├── views.py
│   ├── serializers.py
│   ├── services.py
│   ├── tasks.py
│   ├── urls.py
│   └── tests/
├── core/
│   ├── permissions.py
│   ├── exceptions.py
│   └── middleware.py
└── tests/

Integration Points

• Receives from: main-orchestrator
• Provides APIs to: frontend-orchestrator
• Communicates with: Database, Redis, External APIs
• Outputs to: test-orchestrator

Common Patterns

• Repository Pattern: Data access abstraction
• Service Layer: Business logic encapsulation
• Factory Pattern: Object creation
• Strategy Pattern: Payment processors
• Observer Pattern: Event handling

Anti-Patterns to Avoid

• Fat models (> 200 lines)
• Business logic in views
• Synchronous heavy operations
• N+1 queries
• Missing input validation
• Hardcoded secrets
• Poor error messages