FastAPI to Django Conversion Guide

Status: Active Development Date: 2025-11-30 Purpose: Line-by-line conversion patterns from FastAPI to Django REST Framework based on ADR-007

Import Statements
Request Handling
Model Definitions
Serializers (Pydantic → DRF)
ViewSets (Route Handlers → ViewSets)
Middleware
Authentication
Validation
Async Handling
Database Access
Complete File Examples

1. Import Statements

FastAPI Pattern

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from pydantic import BaseModel, Field, validator
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload
from typing import Optional, List
import httpx

Django Equivalent

# Django core
from django.db import models
from django.contrib.auth.models import AbstractBaseUser, PermissionsMixin
from django.utils import timezone
from django.core.exceptions import ObjectDoesNotExist

# Django REST Framework
from rest_framework import viewsets, status
from rest_framework.decorators import action
from rest_framework.response import Response
from rest_framework.permissions import IsAuthenticated
from rest_framework.authentication import SessionAuthentication

# Custom apps
from apps.tenants.models import TenantModel, Tenant
from apps.tenants.context import get_current_tenant, set_current_tenant
from apps.auth.authentication import JWTAuthentication

# Python standard library
from typing import Optional, List
import uuid

Key Differences

FastAPI	Django	Notes
`from fastapi import FastAPI`	`from django.db import models`	Django uses apps, not single instance
`from pydantic import BaseModel`	`from rest_framework import serializers`	DRF serializers replace Pydantic
`from sqlalchemy import Column`	`from django.db import models`	Django ORM replaces SQLAlchemy
`from fastapi import Depends`	DRF permissions/authentication classes	Dependency injection done differently
`HTTPException`	`rest_framework.exceptions.*`	DRF has specific exception classes

2. Request Handling

FastAPI Pattern

from fastapi import FastAPI, Depends

app = FastAPI()

@app.get("/api/v1/users/{user_id}")
async def get_user(
    user_id: str,
    current_user: User = Depends(get_current_user)
):
    return {"user_id": user_id}

@app.post("/api/v1/users")
async def create_user(
    user_data: UserCreate,
    current_user: User = Depends(get_current_admin)
):
    # Create user logic
    return user_data

Django Equivalent

# apps/users/views.py
from rest_framework import viewsets
from rest_framework.decorators import action
from rest_framework.response import Response
from rest_framework.permissions import IsAuthenticated

class UserViewSet(viewsets.ModelViewSet):
    """User management API."""
    queryset = User.objects.all()
    serializer_class = UserSerializer
    permission_classes = [IsAuthenticated]

    def retrieve(self, request, pk=None):
        """GET /api/v1/users/{pk}/"""
        user = self.get_object()
        serializer = self.get_serializer(user)
        return Response(serializer.data)

    def create(self, request):
        """POST /api/v1/users/"""
        serializer = self.get_serializer(data=request.data)
        serializer.is_valid(raise_exception=True)
        self.perform_create(serializer)
        return Response(serializer.data, status=status.HTTP_201_CREATED)

    @action(detail=False, methods=['get'])
    def me(self, request):
        """GET /api/v1/users/me/"""
        serializer = self.get_serializer(request.user)
        return Response(serializer.data)

Key Differences

FastAPI	Django	Notes
`@app.get("/path")`	`def retrieve(self, request, pk=None)`	DRF uses HTTP method names
`@app.post("/path")`	`def create(self, request)`	ViewSet methods map to HTTP verbs
`Depends(get_current_user)`	`self.request.user`	User available on `request` object
Path params as function args	`pk` parameter in method signature	DRF handles URL routing
Custom endpoints	`@action(detail=True/False)`	Custom actions use decorator

3. Model Definitions

FastAPI Pattern (SQLAlchemy)

from sqlalchemy import Column, String, DateTime, ForeignKey, Boolean
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import relationship
from datetime import datetime
import uuid

class User(Base):
    __tablename__ = "users"

    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    tenant_id = Column(UUID(as_uuid=True), ForeignKey("tenants.id"), nullable=False)
    email = Column(String(255), nullable=False)
    username = Column(String(150), nullable=False)
    password_hash = Column(String(255), nullable=False)
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime, default=datetime.utcnow)

    tenant = relationship("Tenant", back_populates="users")
    license_sessions = relationship("LicenseSession", back_populates="user")

Django Equivalent (From ADR-007)

# apps/users/models.py
from django.contrib.auth.models import AbstractBaseUser, PermissionsMixin
from django.db import models
from apps.tenants.models import TenantModel, Tenant
import uuid

class UserRole(models.TextChoices):
    OWNER = 'owner', 'Owner'
    ADMIN = 'admin', 'Admin'
    MEMBER = 'member', 'Member'
    VIEWER = 'viewer', 'Viewer'

class User(TenantModel, AbstractBaseUser, PermissionsMixin):
    id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)

    # Authentication
    email = models.EmailField(max_length=255)
    username = models.CharField(max_length=150)
    password = models.CharField(max_length=255)

    # Profile
    full_name = models.CharField(max_length=255, blank=True)

    # Status
    is_active = models.BooleanField(default=True)
    is_staff = models.BooleanField(default=False)

    # Tenant role
    role = models.CharField(
        max_length=50,
        choices=UserRole.choices,
        default=UserRole.MEMBER
    )

    # Timestamps
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)
    last_login = models.DateTimeField(null=True, blank=True)

    USERNAME_FIELD = 'email'
    REQUIRED_FIELDS = ['username']

    class Meta:
        db_table = 'users'
        unique_together = [
            ('tenant', 'email'),
            ('tenant', 'username'),
        ]
        indexes = [
            models.Index(fields=['tenant', 'email']),
            models.Index(fields=['tenant', 'username']),
        ]

    def __str__(self):
        return f"{self.email} ({self.tenant.name})"

    @property
    def is_owner(self):
        return self.role == UserRole.OWNER

    @property
    def is_admin(self):
        return self.role in [UserRole.OWNER, UserRole.ADMIN]

Key Differences

SQLAlchemy	Django ORM	Notes
`Column(UUID(...), primary_key=True)`	`models.UUIDField(primary_key=True)`	Django has native UUID support
`Column(String(255))`	`models.CharField(max_length=255)`	Different syntax for string fields
`Column(Boolean, default=True)`	`models.BooleanField(default=True)`	Similar patterns
`Column(DateTime)`	`models.DateTimeField()`	DateTime fields similar
`ForeignKey("tenants.id")`	`models.ForeignKey(Tenant, on_delete=...)`	Django requires `on_delete`
`relationship("Tenant")`	Reverse relation via `related_name`	Django does reverse automatically
`__tablename__`	`class Meta: db_table`	Table name in Meta class
No `__str__`	`def __str__(self)`	Django best practice

Tenant-Scoped Model (From ADR-007)

# apps/tenants/models.py
from django_multitenant.models import TenantModel as BaseTenantModel

class TenantModel(BaseTenantModel):
    """
    Base class for all tenant-scoped models.

    Automatically filters queries to current tenant and enforces
    tenant FK on all operations.
    """

    tenant = models.ForeignKey(
        Tenant,
        on_delete=models.CASCADE,
        related_name='%(class)ss',  # e.g., tenant.users, tenant.projects
        db_index=True
    )

    class Meta:
        abstract = True

    def save(self, *args, **kwargs):
        # Ensure tenant is set from context if not provided
        if not self.tenant_id:
            from .context import get_current_tenant
            current_tenant = get_current_tenant()
            if current_tenant:
                self.tenant_id = current_tenant.id
            else:
                raise ValueError(
                    f"Cannot save {self.__class__.__name__} without tenant context. "
                    "Set tenant explicitly or use TenantMiddleware."
                )
        super().save(*args, **kwargs)

Usage:

# All tenant-scoped models inherit from TenantModel
class LicenseSession(TenantModel):
    # Automatically gets tenant FK and filtering
    user = models.ForeignKey(User, on_delete=models.CASCADE)
    session_token = models.CharField(max_length=255, unique=True)
    # ...

4. Serializers (Pydantic → DRF)

FastAPI Pattern (Pydantic)

from pydantic import BaseModel, Field, validator
from typing import Optional
from datetime import datetime
import uuid

class UserBase(BaseModel):
    email: str = Field(..., max_length=255)
    username: str = Field(..., max_length=150)
    full_name: Optional[str] = None

    @validator('email')
    def validate_email(cls, v):
        if '@' not in v:
            raise ValueError('Invalid email address')
        return v.lower()

class UserCreate(UserBase):
    password: str = Field(..., min_length=8)

class UserResponse(UserBase):
    id: uuid.UUID
    tenant_id: uuid.UUID
    is_active: bool
    role: str
    created_at: datetime

    class Config:
        from_attributes = True  # Pydantic v2 (was orm_mode)

Django Equivalent (From ADR-007 Pattern)

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

class UserSerializer(serializers.ModelSerializer):
    """Base user serializer."""

    class Meta:
        model = User
        fields = [
            'id', 'email', 'username', 'full_name',
            'is_active', 'role', 'created_at', 'updated_at'
        ]
        read_only_fields = ['id', 'created_at', 'updated_at']

    def validate_email(self, value):
        """Validate and normalize email."""
        if '@' not in value:
            raise serializers.ValidationError("Invalid email address")
        return value.lower()

class UserCreateSerializer(serializers.ModelSerializer):
    """Serializer for user creation."""
    password = serializers.CharField(
        write_only=True,
        required=True,
        style={'input_type': 'password'},
        validators=[validate_password]
    )
    password_confirm = serializers.CharField(
        write_only=True,
        required=True,
        style={'input_type': 'password'}
    )

    class Meta:
        model = User
        fields = [
            'email', 'username', 'full_name',
            'password', 'password_confirm', 'role'
        ]

    def validate(self, attrs):
        """Validate password confirmation."""
        if attrs.get('password') != attrs.get('password_confirm'):
            raise serializers.ValidationError({
                "password_confirm": "Passwords do not match"
            })
        attrs.pop('password_confirm')  # Remove from validated data
        return attrs

    def create(self, validated_data):
        """Create user with hashed password."""
        password = validated_data.pop('password')
        user = User(**validated_data)
        user.set_password(password)  # Hash password
        user.save()
        return user

class UserDetailSerializer(UserSerializer):
    """Detailed user serializer with related data."""
    tenant_name = serializers.CharField(source='tenant.name', read_only=True)
    license_sessions_count = serializers.SerializerMethodField()

    class Meta(UserSerializer.Meta):
        fields = UserSerializer.Meta.fields + [
            'tenant_name', 'license_sessions_count', 'last_login'
        ]

    def get_license_sessions_count(self, obj):
        """Count active license sessions."""
        return obj.license_sessions.filter(status='active').count()

Key Differences

Pydantic	Django REST Framework	Notes
`class UserBase(BaseModel)`	`class UserSerializer(serializers.ModelSerializer)`	ModelSerializer auto-generates fields
`Field(..., max_length=255)`	`class Meta: model = User`	DRF infers from model
`@validator('field')`	`def validate_field(self, value)`	DRF validation method naming
`class Config: from_attributes = True`	Not needed	DRF handles ORM automatically
Multiple schemas (Base, Create, Response)	Multiple serializers with inheritance	Similar pattern
`Optional[str] = None`	`required=False, allow_null=True`	DRF explicit about nullability

5. ViewSets (Route Handlers → ViewSets)

FastAPI Pattern

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession

router = APIRouter(prefix="/api/v1/users", tags=["users"])

@router.get("/", response_model=List[UserResponse])
async def list_users(
    skip: int = 0,
    limit: int = 50,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_admin)
):
    result = await db.execute(
        select(User)
        .filter(User.tenant_id == current_user.tenant_id)
        .offset(skip)
        .limit(limit)
    )
    users = result.scalars().all()
    return users

@router.get("/{user_id}", response_model=UserResponse)
async def get_user(
    user_id: uuid.UUID,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user)
):
    result = await db.execute(
        select(User)
        .filter(User.id == user_id)
        .filter(User.tenant_id == current_user.tenant_id)
    )
    user = result.scalar_one_or_none()

    if not user:
        raise HTTPException(status_code=404, detail="User not found")

    return user

@router.post("/", response_model=UserResponse, status_code=201)
async def create_user(
    user_data: UserCreate,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_admin)
):
    user = User(**user_data.dict(), tenant_id=current_user.tenant_id)
    db.add(user)
    await db.commit()
    await db.refresh(user)
    return user

@router.delete("/{user_id}", status_code=204)
async def delete_user(
    user_id: uuid.UUID,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_admin)
):
    result = await db.execute(
        select(User)
        .filter(User.id == user_id)
        .filter(User.tenant_id == current_user.tenant_id)
    )
    user = result.scalar_one_or_none()

    if not user:
        raise HTTPException(status_code=404, detail="User not found")

    await db.delete(user)
    await db.commit()

Django Equivalent (From ADR-007)

# 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
from apps.users.models import User
from apps.users.serializers import (
    UserSerializer,
    UserCreateSerializer,
    UserDetailSerializer
)
from apps.tenants.context import get_current_tenant

class UserViewSet(viewsets.ModelViewSet):
    """
    User management API.

    Endpoints:
    - GET    /api/v1/users/         List users
    - POST   /api/v1/users/         Create user
    - GET    /api/v1/users/{id}/    Get user details
    - PUT    /api/v1/users/{id}/    Update user
    - PATCH  /api/v1/users/{id}/    Partial update
    - DELETE /api/v1/users/{id}/    Delete user
    - GET    /api/v1/users/me/      Get current user
    """

    queryset = User.objects.all()
    permission_classes = [IsAuthenticated]

    def get_serializer_class(self):
        """Return appropriate serializer based on action."""
        if self.action == 'create':
            return UserCreateSerializer
        elif self.action == 'retrieve':
            return UserDetailSerializer
        return UserSerializer

    def get_queryset(self):
        """Filter queryset to current tenant."""
        queryset = super().get_queryset()

        # Queryset is already filtered by RLS + django-multitenant,
        # but we add explicit filter for clarity
        tenant = get_current_tenant()
        if tenant:
            queryset = queryset.filter(tenant=tenant)

        # Additional filtering based on user role
        user = self.request.user

        if user.is_owner or user.is_admin:
            # Owners and admins see all users in tenant
            return queryset
        else:
            # Members see only themselves
            return queryset.filter(id=user.id)

    def list(self, request):
        """
        GET /api/v1/users/?limit=50&offset=0

        Query Parameters:
        - limit: Page size (default 50)
        - offset: Page offset (default 0)
        """
        queryset = self.filter_queryset(self.get_queryset())

        page = self.paginate_queryset(queryset)
        if page is not None:
            serializer = self.get_serializer(page, many=True)
            return self.get_paginated_response(serializer.data)

        serializer = self.get_serializer(queryset, many=True)
        return Response(serializer.data)

    def retrieve(self, request, pk=None):
        """GET /api/v1/users/{pk}/"""
        user = self.get_object()
        serializer = self.get_serializer(user)
        return Response(serializer.data)

    def create(self, request):
        """POST /api/v1/users/"""
        serializer = self.get_serializer(data=request.data)
        serializer.is_valid(raise_exception=True)
        self.perform_create(serializer)

        # Use detail serializer for response
        response_serializer = UserDetailSerializer(serializer.instance)
        return Response(
            response_serializer.data,
            status=status.HTTP_201_CREATED
        )

    def perform_create(self, serializer):
        """Set tenant on create."""
        tenant = get_current_tenant()
        if not tenant:
            from rest_framework.exceptions import PermissionDenied
            raise PermissionDenied("No tenant context available")

        serializer.save(tenant=tenant)

    def update(self, request, pk=None):
        """PUT /api/v1/users/{pk}/"""
        user = self.get_object()
        serializer = self.get_serializer(user, data=request.data)
        serializer.is_valid(raise_exception=True)
        serializer.save()
        return Response(serializer.data)

    def partial_update(self, request, pk=None):
        """PATCH /api/v1/users/{pk}/"""
        user = self.get_object()
        serializer = self.get_serializer(user, data=request.data, partial=True)
        serializer.is_valid(raise_exception=True)
        serializer.save()
        return Response(serializer.data)

    def destroy(self, request, pk=None):
        """DELETE /api/v1/users/{pk}/"""
        user = self.get_object()
        user.delete()
        return Response(status=status.HTTP_204_NO_CONTENT)

    @action(detail=False, methods=['get'])
    def me(self, request):
        """
        GET /api/v1/users/me/

        Get current authenticated user.
        """
        serializer = UserDetailSerializer(request.user)
        return Response(serializer.data)

    @action(detail=True, methods=['post'])
    def deactivate(self, request, pk=None):
        """
        POST /api/v1/users/{pk}/deactivate/

        Deactivate a user account.
        """
        user = self.get_object()
        user.is_active = False
        user.save(update_fields=['is_active'])
        return Response({'status': 'user deactivated'})

URL Configuration (NEW)

# apps/users/urls.py
from django.urls import path, include
from rest_framework.routers import DefaultRouter
from apps.users.views import UserViewSet

router = DefaultRouter()
router.register(r'users', UserViewSet, basename='user')

urlpatterns = [
    path('api/v1/', include(router.urls)),
]

Key Differences

FastAPI	Django REST Framework	Notes
`@router.get("/")`	`def list(self, request)`	DRF ViewSet method
`@router.get("/{id}")`	`def retrieve(self, request, pk=None)`	pk = primary key
`@router.post("/")`	`def create(self, request)`	Standard CRUD method
`@router.delete("/{id}")`	`def destroy(self, request, pk=None)`	Destroy, not delete
Manual pagination	`self.paginate_queryset()`	DRF handles pagination
`Depends(get_current_user)`	`self.request.user`	User on request
`Depends(get_db)`	Django ORM (no injection)	No DB session needed
Custom endpoint	`@action(detail=True/False)`	Custom actions
Explicit 404 raise	`self.get_object()`	DRF raises 404 automatically

6. Middleware

FastAPI Pattern

from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware

class TenantMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        # Extract tenant from user or headers
        if request.user and hasattr(request.user, 'tenant_id'):
            request.state.tenant_id = request.user.tenant_id
        else:
            request.state.tenant_id = None

        response = await call_next(request)
        return response

Django Equivalent (From ADR-007)

# apps/tenants/middleware.py
from django.utils.deprecation import MiddlewareMixin
from django.http import JsonResponse
from .context import set_current_tenant, clear_current_tenant

class TenantMiddleware(MiddlewareMixin):
    """
    Middleware to set current tenant from authenticated user.

    CRITICAL: This middleware must come AFTER AuthenticationMiddleware
    in the MIDDLEWARE setting.
    """

    def process_request(self, request):
        """Set tenant from authenticated user."""
        clear_current_tenant()

        if request.user and request.user.is_authenticated:
            # User has tenant FK, set it as current
            if hasattr(request.user, 'tenant'):
                set_current_tenant(request.user.tenant)
                request.tenant = request.user.tenant
            else:
                # Superuser or staff without tenant (for admin)
                request.tenant = None
        else:
            request.tenant = None

    def process_response(self, request, response):
        """Clear tenant context after request."""
        clear_current_tenant()
        return response

    def process_exception(self, request, exception):
        """Clear tenant context on exception."""
        clear_current_tenant()
        return None

Tenant Context Module (From ADR-007)

# apps/tenants/context.py
from contextvars import ContextVar
from typing import Optional
from .models import Tenant

# Thread-safe context variable for current tenant
_current_tenant: ContextVar[Optional[Tenant]] = ContextVar('current_tenant', default=None)

def set_current_tenant(tenant: Optional[Tenant]) -> None:
    """
    Set the current tenant for this context.

    This is used by TenantMiddleware to set tenant from authenticated user.
    """
    _current_tenant.set(tenant)

    # Also set PostgreSQL session variable for RLS
    if tenant:
        from django.db import connection
        with connection.cursor() as cursor:
            cursor.execute("SELECT set_current_tenant(%s)", [str(tenant.id)])

def get_current_tenant() -> Optional[Tenant]:
    """Get the current tenant from context."""
    return _current_tenant.get()

def clear_current_tenant() -> None:
    """Clear the current tenant context."""
    _current_tenant.set(None)

    # Clear PostgreSQL session variable
    from django.db import connection
    with connection.cursor() as cursor:
        cursor.execute("SELECT clear_current_tenant()")

class tenant_context:
    """
    Context manager for temporarily setting tenant context.

    Usage:
        with tenant_context(my_tenant):
            # All queries here are scoped to my_tenant
            users = User.objects.all()
    """

    def __init__(self, tenant: Optional[Tenant]):
        self.tenant = tenant
        self.previous_tenant = None

    def __enter__(self):
        self.previous_tenant = get_current_tenant()
        set_current_tenant(self.tenant)
        return self.tenant

    def __exit__(self, exc_type, exc_val, exc_tb):
        set_current_tenant(self.previous_tenant)

Settings Configuration (From ADR-007)

# settings/base.py

MIDDLEWARE = [
    'django.middleware.security.SecurityMiddleware',
    'corsheaders.middleware.CorsMiddleware',
    'django.contrib.sessions.middleware.SessionMiddleware',
    'django.middleware.common.CommonMiddleware',
    'django.middleware.csrf.CsrfViewMiddleware',
    'django.contrib.auth.middleware.AuthenticationMiddleware',

    # Tenant middleware MUST come after authentication
    'apps.tenants.middleware.TenantMiddleware',

    'django.contrib.messages.middleware.MessageMiddleware',
    'django.middleware.clickjacking.XFrameOptionsMiddleware',
]

Key Differences

FastAPI	Django	Notes
`BaseHTTPMiddleware`	`MiddlewareMixin`	Django middleware pattern
`async def dispatch()`	`process_request()`, `process_response()`	Django sync middleware
`request.state.tenant_id`	`request.tenant`	Django adds to request directly
`await call_next(request)`	Return from `process_request()` continues chain	Different flow
No cleanup	`process_response()` for cleanup	Django has explicit cleanup

7. Authentication

FastAPI Pattern

from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
import jwt

security = HTTPBearer()

async def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(security),
    db: AsyncSession = Depends(get_db)
) -> User:
    token = credentials.credentials

    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
        user_id = payload.get("sub")
    except jwt.JWTError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid authentication credentials"
        )

    result = await db.execute(select(User).filter(User.id == user_id))
    user = result.scalar_one_or_none()

    if not user:
        raise HTTPException(status_code=401, detail="User not found")

    return user

async def get_current_admin(
    current_user: User = Depends(get_current_user)
) -> User:
    if current_user.role not in ['owner', 'admin']:
        raise HTTPException(status_code=403, detail="Insufficient permissions")
    return current_user

Django Equivalent

# apps/auth/authentication.py
from rest_framework import authentication
from rest_framework import exceptions
from apps.users.models import User
import jwt
from django.conf import settings

class JWTAuthentication(authentication.BaseAuthentication):
    """
    JWT token authentication for API requests.

    Expects header: Authorization: Bearer <token>
    """

    def authenticate(self, request):
        auth_header = request.META.get('HTTP_AUTHORIZATION', '')

        if not auth_header.startswith('Bearer '):
            return None

        token = auth_header.split(' ')[1]

        try:
            payload = jwt.decode(
                token,
                settings.SECRET_KEY,
                algorithms=["HS256"]
            )
            user_id = payload.get("sub")
        except jwt.ExpiredSignatureError:
            raise exceptions.AuthenticationFailed('Token has expired')
        except jwt.JWTError:
            raise exceptions.AuthenticationFailed('Invalid token')

        try:
            user = User.objects.get(id=user_id)
        except User.DoesNotExist:
            raise exceptions.AuthenticationFailed('User not found')

        if not user.is_active:
            raise exceptions.AuthenticationFailed('User is inactive')

        return (user, token)

    def authenticate_header(self, request):
        return 'Bearer'

Permission Classes

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

class IsOwnerOrAdmin(permissions.BasePermission):
    """
    Permission check for owner or admin role.
    """

    def has_permission(self, request, view):
        return (
            request.user and
            request.user.is_authenticated and
            request.user.role in ['owner', 'admin']
        )

class IsTenantMember(permissions.BasePermission):
    """
    Permission check for tenant membership.
    """

    def has_object_permission(self, request, view, obj):
        if not request.user or not request.user.is_authenticated:
            return False

        # Check if object has tenant and user is in same tenant
        if hasattr(obj, 'tenant'):
            return obj.tenant_id == request.user.tenant_id

        return False

Usage in ViewSets

from rest_framework.permissions import IsAuthenticated
from apps.auth.permissions import IsOwnerOrAdmin

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

    def get_permissions(self):
        """
        Override to use different permissions per action.
        """
        if self.action in ['create', 'destroy']:
            # Only owners/admins can create or delete users
            return [IsOwnerOrAdmin()]
        return [IsAuthenticated()]

Settings Configuration

# settings/base.py

REST_FRAMEWORK = {
    'DEFAULT_AUTHENTICATION_CLASSES': [
        'rest_framework.authentication.SessionAuthentication',
        'apps.auth.authentication.JWTAuthentication',
    ],
    'DEFAULT_PERMISSION_CLASSES': [
        'rest_framework.permissions.IsAuthenticated',
    ],
}

Key Differences

FastAPI	Django REST Framework	Notes
`Depends(security)`	Authentication class	DRF uses classes
`Depends(get_current_user)`	`request.user`	User auto-populated
`Depends(get_current_admin)`	Permission class	Permissions separate
Dependency injection	Settings configuration	DRF configured globally
Function-based auth	Class-based auth	DRF uses classes

8. Validation

FastAPI Pattern (Pydantic)

from pydantic import BaseModel, validator, root_validator, Field
from typing import Optional
import re

class UserCreate(BaseModel):
    email: str = Field(..., max_length=255)
    password: str = Field(..., min_length=8)
    password_confirm: str

    @validator('email')
    def validate_email(cls, v):
        email_regex = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
        if not re.match(email_regex, v):
            raise ValueError('Invalid email format')
        return v.lower()

    @validator('password')
    def validate_password_strength(cls, v):
        if not any(char.isdigit() for char in v):
            raise ValueError('Password must contain at least one digit')
        if not any(char.isupper() for char in v):
            raise ValueError('Password must contain at least one uppercase letter')
        return v

    @root_validator
    def validate_passwords_match(cls, values):
        password = values.get('password')
        password_confirm = values.get('password_confirm')

        if password != password_confirm:
            raise ValueError('Passwords do not match')

        return values

Django Equivalent

# apps/users/serializers.py
from rest_framework import serializers
from django.contrib.auth.password_validation import validate_password
from django.core.exceptions import ValidationError as DjangoValidationError
import re

class UserCreateSerializer(serializers.ModelSerializer):
    password = serializers.CharField(
        write_only=True,
        required=True,
        style={'input_type': 'password'},
        validators=[validate_password]  # Django built-in password validation
    )
    password_confirm = serializers.CharField(
        write_only=True,
        required=True,
        style={'input_type': 'password'}
    )

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

    def validate_email(self, value):
        """
        Validate email format and uniqueness.

        Field-level validation - called automatically for 'email' field.
        """
        email_regex = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
        if not re.match(email_regex, value):
            raise serializers.ValidationError('Invalid email format')

        # Check uniqueness within tenant
        from apps.tenants.context import get_current_tenant
        tenant = get_current_tenant()

        if tenant and User.objects.filter(
            tenant=tenant,
            email=value.lower()
        ).exists():
            raise serializers.ValidationError('Email already exists in this tenant')

        return value.lower()

    def validate_password(self, value):
        """
        Additional password validation beyond Django's built-in.

        Field-level validation for 'password'.
        """
        if not any(char.isdigit() for char in value):
            raise serializers.ValidationError(
                'Password must contain at least one digit'
            )

        if not any(char.isupper() for char in value):
            raise serializers.ValidationError(
                'Password must contain at least one uppercase letter'
            )

        return value

    def validate(self, attrs):
        """
        Object-level validation - access to all fields.

        Called after all field-level validators.
        """
        password = attrs.get('password')
        password_confirm = attrs.get('password_confirm')

        if password != password_confirm:
            raise serializers.ValidationError({
                "password_confirm": "Passwords do not match"
            })

        # Remove password_confirm from validated data
        attrs.pop('password_confirm')

        return attrs

    def create(self, validated_data):
        """Create user with proper password hashing."""
        password = validated_data.pop('password')
        user = User(**validated_data)
        user.set_password(password)  # Hash password
        user.save()
        return user

Custom Validators (Reusable)

# apps/common/validators.py
from django.core.exceptions import ValidationError
import re

def validate_username_format(value):
    """
    Validate username contains only allowed characters.

    Usage in model:
        username = models.CharField(validators=[validate_username_format])
    """
    username_regex = r'^[a-zA-Z0-9_-]+$'
    if not re.match(username_regex, value):
        raise ValidationError(
            'Username can only contain letters, numbers, hyphens, and underscores'
        )

def validate_phone_number(value):
    """Validate phone number format."""
    phone_regex = r'^\+?1?\d{9,15}$'
    if not re.match(phone_regex, value):
        raise ValidationError('Invalid phone number format')

Model-Level Validation

# apps/users/models.py
from django.core.exceptions import ValidationError

class User(TenantModel):
    # ... fields ...

    def clean(self):
        """
        Model-level validation.

        Called before save() when using Django forms or admin.
        For API, use serializer validation instead.
        """
        super().clean()

        # Example: Validate email domain for specific tenants
        if self.tenant and self.tenant.slug == 'enterprise-client':
            if not self.email.endswith('@enterprise-client.com'):
                raise ValidationError({
                    'email': 'Email must be from @enterprise-client.com domain'
                })

    def save(self, *args, **kwargs):
        """Override save to run validation."""
        if not kwargs.get('skip_validation'):
            self.full_clean()  # Runs clean() and field validators
        super().save(*args, **kwargs)

Key Differences

Pydantic	Django REST Framework	Notes
`@validator('field')`	`def validate_field(self, value)`	Method naming convention
`@root_validator`	`def validate(self, attrs)`	Object-level validation
Field(..., min_length=8)	`validators=[validate_password]`	DRF uses validator functions
Raises `ValueError`	Raises `serializers.ValidationError`	Different exception types
Validation in schema	Validation in serializer	Similar separation

9. Async Handling

FastAPI Pattern (Native Async)

from fastapi import FastAPI, Depends
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select

app = FastAPI()

@app.get("/api/v1/users/{user_id}")
async def get_user(
    user_id: str,
    db: AsyncSession = Depends(get_async_db)
):
    result = await db.execute(
        select(User).filter(User.id == user_id)
    )
    user = result.scalar_one_or_none()
    return user

@app.post("/api/v1/users/{user_id}/send-email")
async def send_email(user_id: str, db: AsyncSession = Depends(get_async_db)):
    # Async external API call
    async with httpx.AsyncClient() as client:
        response = await client.post(
            "https://api.sendgrid.com/v3/mail/send",
            json={...}
        )

    return {"status": "sent"}

Django Equivalent (Django 4.2+ Async Views)

# apps/users/views.py
from rest_framework import viewsets
from rest_framework.decorators import action
from rest_framework.response import Response
from asgiref.sync import sync_to_async
import httpx

class UserViewSet(viewsets.ModelViewSet):
    queryset = User.objects.all()
    serializer_class = UserSerializer

    def retrieve(self, request, pk=None):
        """
        Synchronous view (default).

        Django ORM is synchronous by default.
        """
        user = self.get_object()
        serializer = self.get_serializer(user)
        return Response(serializer.data)

    @action(detail=True, methods=['post'])
    async def send_email(self, request, pk=None):
        """
        Async view for I/O-bound operations.

        Use async for:
        - External API calls
        - File I/O
        - Network requests
        """
        # Wrap synchronous ORM call in sync_to_async
        user = await sync_to_async(self.get_object)()

        # Async external API call
        async with httpx.AsyncClient() as client:
            response = await client.post(
                "https://api.sendgrid.com/v3/mail/send",
                json={
                    "to": user.email,
                    "subject": "Welcome",
                    "body": "Welcome to CODITECT!"
                }
            )

        return Response({"status": "sent", "message_id": response.json()['id']})

Async ORM (Django 4.1+)

# apps/licenses/views.py
from rest_framework import viewsets
from asgiref.sync import sync_to_async
from django.db import models

class LicenseSessionViewSet(viewsets.ModelViewSet):
    queryset = LicenseSession.objects.all()
    serializer_class = LicenseSessionSerializer

    @action(detail=False, methods=['post'])
    async def validate_bulk(self, request):
        """
        Async bulk validation with async ORM.

        Django 4.1+ supports async ORM queries.
        """
        session_tokens = request.data.get('session_tokens', [])

        # Async ORM query (Django 4.1+)
        sessions = [
            session async for session in
            LicenseSession.objects.filter(
                session_token__in=session_tokens
            ).aiterator()
        ]

        results = []
        for session in sessions:
            is_valid = await sync_to_async(session.validate)()
            results.append({
                "session_token": session.session_token,
                "is_valid": is_valid
            })

        return Response({"results": results})

Background Tasks (Celery)

# apps/licenses/tasks.py
from celery import shared_task
from apps.licenses.models import LicenseSession
from django.utils import timezone

@shared_task
def cleanup_expired_sessions():
    """
    Celery task for background processing.

    Use for:
    - Long-running operations
    - Scheduled tasks
    - Batch processing
    """
    expired_sessions = LicenseSession.objects.filter(
        status='active',
        expires_at__lt=timezone.now()
    )

    count = expired_sessions.update(status='expired')
    return f"Expired {count} sessions"

# In views.py
from apps.licenses.tasks import cleanup_expired_sessions

@action(detail=False, methods=['post'])
def trigger_cleanup(self, request):
    """Trigger background task."""
    task = cleanup_expired_sessions.delay()
    return Response({"task_id": task.id, "status": "queued"})

Settings Configuration

# settings/base.py

# Celery configuration
CELERY_BROKER_URL = os.getenv('REDIS_URL', 'redis://localhost:6379/0')
CELERY_RESULT_BACKEND = CELERY_BROKER_URL
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_ACCEPT_CONTENT = ['json']
CELERY_TIMEZONE = 'UTC'

# Celery Beat (scheduled tasks)
from celery.schedules import crontab

CELERY_BEAT_SCHEDULE = {
    'cleanup-expired-sessions': {
        'task': 'apps.licenses.tasks.cleanup_expired_sessions',
        'schedule': crontab(minute='*/5'),  # Every 5 minutes
    },
}

Key Differences

FastAPI	Django	Notes
Native async	Async views in DRF (4.2+)	Django added async support
`async def endpoint()`	`async def action()`	Similar syntax
AsyncSession (SQLAlchemy)	sync_to_async wrapper	Django ORM mostly sync
Direct async/await	sync_to_async for ORM	Wrapper needed
Background tasks inline	Celery tasks	Django uses Celery

10. Database Access

FastAPI Pattern (SQLAlchemy Async)

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker, selectinload
from sqlalchemy import select, update, delete

# Engine setup
engine = create_async_engine(
    "postgresql+asyncpg://user:pass@localhost/db",
    echo=True
)

AsyncSessionLocal = sessionmaker(
    engine, class_=AsyncSession, expire_on_commit=False
)

async def get_db():
    async with AsyncSessionLocal() as session:
        yield session

# Usage in endpoints
@app.get("/api/v1/users/{user_id}")
async def get_user(
    user_id: str,
    db: AsyncSession = Depends(get_db)
):
    # SELECT query with relationships
    result = await db.execute(
        select(User)
        .options(selectinload(User.license_sessions))
        .filter(User.id == user_id)
    )
    user = result.scalar_one_or_none()
    return user

@app.post("/api/v1/users")
async def create_user(
    user_data: UserCreate,
    db: AsyncSession = Depends(get_db)
):
    # INSERT
    user = User(**user_data.dict())
    db.add(user)
    await db.commit()
    await db.refresh(user)
    return user

@app.put("/api/v1/users/{user_id}")
async def update_user(
    user_id: str,
    user_data: UserUpdate,
    db: AsyncSession = Depends(get_db)
):
    # UPDATE
    await db.execute(
        update(User)
        .where(User.id == user_id)
        .values(**user_data.dict(exclude_unset=True))
    )
    await db.commit()

    # Refresh to get updated data
    result = await db.execute(select(User).filter(User.id == user_id))
    user = result.scalar_one()
    return user

@app.delete("/api/v1/users/{user_id}")
async def delete_user(user_id: str, db: AsyncSession = Depends(get_db)):
    # DELETE
    await db.execute(delete(User).where(User.id == user_id))
    await db.commit()

Django Equivalent (Django ORM)

# apps/users/views.py
from rest_framework import viewsets
from rest_framework.response import Response
from apps.users.models import User
from apps.tenants.context import get_current_tenant
from django.db.models import Prefetch

class UserViewSet(viewsets.ModelViewSet):
    queryset = User.objects.all()
    serializer_class = UserSerializer

    def get_queryset(self):
        """
        Get queryset with tenant filtering.

        Django ORM is synchronous by default.
        No database session injection needed.
        """
        queryset = super().get_queryset()

        # Tenant filtering (automatic via RLS + django-multitenant)
        tenant = get_current_tenant()
        if tenant:
            queryset = queryset.filter(tenant=tenant)

        # Prefetch related data (like SQLAlchemy selectinload)
        queryset = queryset.select_related('tenant')  # JOIN
        queryset = queryset.prefetch_related(  # Separate query
            Prefetch(
                'license_sessions',
                queryset=LicenseSession.objects.filter(status='active')
            )
        )

        return queryset

    def retrieve(self, request, pk=None):
        """
        SELECT query.

        Django ORM:
        - No session management needed
        - Automatic tenant filtering via RLS
        - get_object() handles 404
        """
        user = self.get_object()  # SELECT * FROM users WHERE id = pk
        serializer = self.get_serializer(user)
        return Response(serializer.data)

    def create(self, request):
        """
        INSERT query.

        Django ORM:
        - Validation via serializer
        - save() commits automatically
        - No explicit commit() needed
        """
        serializer = self.get_serializer(data=request.data)
        serializer.is_valid(raise_exception=True)

        # INSERT INTO users (...)
        self.perform_create(serializer)

        return Response(serializer.data, status=status.HTTP_201_CREATED)

    def perform_create(self, serializer):
        """Set tenant automatically."""
        tenant = get_current_tenant()
        serializer.save(tenant=tenant)  # Calls user.save()

    def update(self, request, pk=None):
        """
        UPDATE query.

        Django ORM:
        - get_object() fetches current state
        - save() updates changed fields only
        - No explicit commit
        """
        user = self.get_object()  # SELECT for locking
        serializer = self.get_serializer(user, data=request.data)
        serializer.is_valid(raise_exception=True)
        serializer.save()  # UPDATE users SET ... WHERE id = pk

        return Response(serializer.data)

    def partial_update(self, request, pk=None):
        """PATCH - partial update."""
        user = self.get_object()
        serializer = self.get_serializer(user, data=request.data, partial=True)
        serializer.is_valid(raise_exception=True)
        serializer.save()  # UPDATE only provided fields

        return Response(serializer.data)

    def destroy(self, request, pk=None):
        """
        DELETE query.

        Django ORM:
        - get_object() fetches for deletion
        - delete() commits automatically
        """
        user = self.get_object()
        user.delete()  # DELETE FROM users WHERE id = pk

        return Response(status=status.HTTP_204_NO_CONTENT)

# Manual ORM queries (outside ViewSets)
def example_queries():
    """Examples of Django ORM queries."""

    # SELECT all
    users = User.objects.all()

    # SELECT with filter
    active_users = User.objects.filter(is_active=True)

    # SELECT with multiple conditions
    from django.db.models import Q
    users = User.objects.filter(
        Q(role='owner') | Q(role='admin'),
        is_active=True
    )

    # SELECT with ordering
    users = User.objects.filter(is_active=True).order_by('-created_at')

    # SELECT with limit/offset (pagination)
    users = User.objects.all()[10:20]  # LIMIT 10 OFFSET 10

    # SELECT with aggregation
    from django.db.models import Count
    tenant = Tenant.objects.annotate(
        user_count=Count('users')
    ).first()

    # SELECT with relationships (JOIN)
    users = User.objects.select_related('tenant').all()

    # SELECT with reverse relationships (separate query)
    users = User.objects.prefetch_related('license_sessions').all()

    # UPDATE (bulk)
    User.objects.filter(is_active=False).update(status='inactive')

    # DELETE (bulk)
    User.objects.filter(status='inactive').delete()

    # Raw SQL (when needed)
    from django.db import connection
    with connection.cursor() as cursor:
        cursor.execute("SELECT * FROM users WHERE tenant_id = %s", [tenant_id])
        rows = cursor.fetchall()

Tenant Context in Queries (From ADR-007)

# Automatic tenant filtering via RLS + django-multitenant
from apps.tenants.context import set_current_tenant, tenant_context

# Method 1: Middleware sets context (automatic in API requests)
# TenantMiddleware sets tenant from request.user.tenant

# Method 2: Manual context setting
tenant = Tenant.objects.get(slug='acme-corp')
set_current_tenant(tenant)

# All queries now automatically filtered to this tenant
users = User.objects.all()  # Only users in 'acme-corp'
projects = Project.objects.all()  # Only projects in 'acme-corp'

# Method 3: Context manager (temporary tenant switch)
with tenant_context(other_tenant):
    # Queries here scoped to other_tenant
    users = User.objects.all()
# Context restored to previous tenant after block

Key Differences

SQLAlchemy Async	Django ORM	Notes
`async with session`	No session management	Django handles automatically
`await db.execute(select(...))`	`Model.objects.filter(...)`	Django query API simpler
`await db.commit()`	`model.save()` commits automatically	No explicit commits
`await db.refresh(obj)`	`obj.refresh_from_db()`	Similar pattern
`selectinload(relationship)`	`prefetch_related('relationship')`	Similar eager loading
Explicit session injection	Global ORM access	No dependency injection
`session.scalar_one()`	`Model.objects.get()`	Different naming

11. Complete File Examples

Example 1: User Management (Complete Conversion)

FastAPI Version (Before)

# api/users.py
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from typing import List
from pydantic import BaseModel, Field
import uuid

router = APIRouter(prefix="/api/v1/users", tags=["users"])

# Pydantic schemas
class UserBase(BaseModel):
    email: str = Field(..., max_length=255)
    username: str = Field(..., max_length=150)
    full_name: str | None = None

class UserCreate(UserBase):
    password: str = Field(..., min_length=8)

class UserResponse(UserBase):
    id: uuid.UUID
    tenant_id: uuid.UUID
    is_active: bool
    role: str
    created_at: datetime

    class Config:
        from_attributes = True

# Endpoints
@router.get("/", response_model=List[UserResponse])
async def list_users(
    skip: int = 0,
    limit: int = 50,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user)
):
    result = await db.execute(
        select(User)
        .filter(User.tenant_id == current_user.tenant_id)
        .offset(skip)
        .limit(limit)
    )
    users = result.scalars().all()
    return users

@router.get("/{user_id}", response_model=UserResponse)
async def get_user(
    user_id: uuid.UUID,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user)
):
    result = await db.execute(
        select(User)
        .filter(User.id == user_id)
        .filter(User.tenant_id == current_user.tenant_id)
    )
    user = result.scalar_one_or_none()

    if not user:
        raise HTTPException(status_code=404, detail="User not found")

    return user

@router.post("/", response_model=UserResponse, status_code=201)
async def create_user(
    user_data: UserCreate,
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_admin)
):
    user = User(**user_data.dict(), tenant_id=current_user.tenant_id)
    user.password_hash = hash_password(user_data.password)
    db.add(user)
    await db.commit()
    await db.refresh(user)
    return user

Django Version (After)

# apps/users/models.py
from django.contrib.auth.models import AbstractBaseUser, PermissionsMixin
from django.db import models
from apps.tenants.models import TenantModel
import uuid

class User(TenantModel, AbstractBaseUser, PermissionsMixin):
    id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
    email = models.EmailField(max_length=255)
    username = models.CharField(max_length=150)
    full_name = models.CharField(max_length=255, blank=True)
    is_active = models.BooleanField(default=True)
    role = models.CharField(max_length=50, default='member')
    created_at = models.DateTimeField(auto_now_add=True)

    USERNAME_FIELD = 'email'
    REQUIRED_FIELDS = ['username']

    class Meta:
        db_table = 'users'
        unique_together = [('tenant', 'email')]

# apps/users/serializers.py
from rest_framework import serializers
from apps.users.models import User

class UserSerializer(serializers.ModelSerializer):
    class Meta:
        model = User
        fields = ['id', 'email', 'username', 'full_name', 'is_active', 'role', 'created_at']
        read_only_fields = ['id', 'created_at']

class UserCreateSerializer(serializers.ModelSerializer):
    password = serializers.CharField(write_only=True, min_length=8)

    class Meta:
        model = User
        fields = ['email', 'username', 'full_name', 'password', 'role']

    def create(self, validated_data):
        password = validated_data.pop('password')
        user = User(**validated_data)
        user.set_password(password)
        user.save()
        return user

# apps/users/views.py
from rest_framework import viewsets
from rest_framework.permissions import IsAuthenticated
from apps.users.models import User
from apps.users.serializers import UserSerializer, UserCreateSerializer
from apps.tenants.context import get_current_tenant

class UserViewSet(viewsets.ModelViewSet):
    queryset = User.objects.all()
    permission_classes = [IsAuthenticated]

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

    def get_queryset(self):
        queryset = super().get_queryset()
        tenant = get_current_tenant()
        if tenant:
            queryset = queryset.filter(tenant=tenant)
        return queryset

    def perform_create(self, serializer):
        tenant = get_current_tenant()
        serializer.save(tenant=tenant)

# apps/users/urls.py
from django.urls import path, include
from rest_framework.routers import DefaultRouter
from apps.users.views import UserViewSet

router = DefaultRouter()
router.register(r'users', UserViewSet, basename='user')

urlpatterns = [
    path('api/v1/', include(router.urls)),
]

Example 2: License Session (With Redis Integration)

Django Version (Full Implementation)

# apps/licenses/models.py
from django.db import models
from apps.tenants.models import TenantModel
from apps.users.models import User
import uuid

class LicenseSession(TenantModel):
    id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
    user = models.ForeignKey(User, on_delete=models.CASCADE, related_name='license_sessions')
    session_token = models.CharField(max_length=255, unique=True, db_index=True)
    machine_id = models.CharField(max_length=255)
    license_type = models.CharField(max_length=50)
    status = models.CharField(max_length=20, default='active')
    created_at = models.DateTimeField(auto_now_add=True)
    expires_at = models.DateTimeField()
    last_validated_at = models.DateTimeField(null=True, blank=True)

    class Meta:
        db_table = 'license_sessions'
        indexes = [
            models.Index(fields=['tenant', 'status']),
            models.Index(fields=['session_token']),
        ]

    @property
    def is_valid(self):
        from django.utils import timezone
        return self.status == 'active' and self.expires_at > timezone.now()

# apps/licenses/services.py
from django.core.cache import cache
from apps.licenses.models import LicenseSession
from django.utils import timezone
import secrets

class LicenseService:
    """Service for license session management with Redis."""

    @staticmethod
    def acquire_license(user, machine_id, license_type='pro'):
        """Acquire a license session with atomic seat counting."""
        from apps.tenants.context import get_current_tenant

        tenant = get_current_tenant()

        # Check seat availability (Redis atomic counter)
        max_seats = tenant.max_users  # From plan tier
        current_seats = cache.get(f'tenant:{tenant.id}:active_seats', 0)

        if current_seats >= max_seats:
            raise ValueError("No available license seats")

        # Create session
        session = LicenseSession.objects.create(
            tenant=tenant,
            user=user,
            session_token=secrets.token_urlsafe(32),
            machine_id=machine_id,
            license_type=license_type,
            expires_at=timezone.now() + timezone.timedelta(hours=24),
            status='active'
        )

        # Increment seat counter (Redis)
        cache.incr(f'tenant:{tenant.id}:active_seats')

        return session

    @staticmethod
    def release_license(session_token):
        """Release a license session."""
        try:
            session = LicenseSession.objects.get(session_token=session_token)
            session.status = 'revoked'
            session.save()

            # Decrement seat counter (Redis)
            cache.decr(f'tenant:{session.tenant.id}:active_seats')

            return True
        except LicenseSession.DoesNotExist:
            return False

# apps/licenses/views.py
from rest_framework import viewsets
from rest_framework.decorators import action
from rest_framework.response import Response
from rest_framework import status
from apps.licenses.models import LicenseSession
from apps.licenses.services import LicenseService

class LicenseSessionViewSet(viewsets.ModelViewSet):
    queryset = LicenseSession.objects.all()

    @action(detail=False, methods=['post'])
    def acquire(self, request):
        """Acquire a license session."""
        machine_id = request.data.get('machine_id')
        license_type = request.data.get('license_type', 'pro')

        try:
            session = LicenseService.acquire_license(
                user=request.user,
                machine_id=machine_id,
                license_type=license_type
            )

            return Response({
                'session_token': session.session_token,
                'expires_at': session.expires_at,
                'license_type': session.license_type
            }, status=status.HTTP_201_CREATED)

        except ValueError as e:
            return Response(
                {'error': str(e)},
                status=status.HTTP_409_CONFLICT
            )

    @action(detail=False, methods=['post'])
    def release(self, request):
        """Release a license session."""
        session_token = request.data.get('session_token')

        if LicenseService.release_license(session_token):
            return Response(status=status.HTTP_204_NO_CONTENT)
        else:
            return Response(
                {'error': 'Session not found'},
                status=status.HTTP_404_NOT_FOUND
            )

Conversion Checklist

Use this checklist when converting each file:

For Each Model File

Replace SQLAlchemy Column() with Django models.*Field()
Replace ForeignKey("table") with models.ForeignKey(Model, on_delete=...)
Replace __tablename__ with class Meta: db_table
Add __str__() method
Inherit from TenantModel for tenant-scoped models
Add class Meta with db_table, unique_together, indexes
Use TextChoices for enums
Replace relationship() with related_name on ForeignKey

For Each Serializer File

Create serializers.py (new file, not in FastAPI)
Replace Pydantic BaseModel with serializers.ModelSerializer
Replace Field(...) with class Meta: model = ...
Replace @validator with def validate_field(self, value)
Replace @root_validator with def validate(self, attrs)
Add create() method for password hashing
Use write_only=True for password fields
Add read_only_fields in Meta

For Each View File

For URLs

Create urls.py (new file)
Use DefaultRouter for ViewSets
Register ViewSets with router
Include router URLs in project URLconf

For Settings

Add apps to INSTALLED_APPS
Configure REST_FRAMEWORK settings
Add middleware (after AuthenticationMiddleware)
Configure DATABASES
Set AUTH_USER_MODEL

For Testing

Replace TestClient with APITestCase
Replace client.get() with self.client.get()
Use force_authenticate(user=user) for auth
Replace assert with self.assertEqual(), etc.
Use TransactionTestCase for RLS testing

Migration Path

Recommended conversion order for ~50 files:

Phase 1: Foundation (Days 1-2)
- Settings configuration
- Tenant model and middleware
- User model and authentication
- Base serializers and permissions
Phase 2: Core Models (Days 3-4)
- License session model
- Project model
- Audit log model
- All tenant-scoped models
Phase 3: API Layer (Days 5-7)
- User ViewSet
- License session ViewSet
- Project ViewSet
- Custom actions and endpoints
Phase 4: Testing (Days 8-9)
- RLS tests
- Tenant isolation tests
- API integration tests
- Performance tests
Phase 5: Deployment (Day 10)
- Docker configuration
- Kubernetes manifests
- CI/CD updates
- Production deployment

Common Pitfalls

1. Forgot to Add Tenant to Model

Wrong:

class Project(models.Model):  # Missing TenantModel!
    name = models.CharField(max_length=255)

Correct:

class Project(TenantModel):  # Inherit from TenantModel
    name = models.CharField(max_length=255)

2. Manual Tenant Filtering Instead of Using Context

Wrong:

def get_queryset(self):
    return User.objects.filter(tenant_id=self.request.user.tenant_id)

Correct:

def get_queryset(self):
    queryset = super().get_queryset()
    tenant = get_current_tenant()
    if tenant:
        queryset = queryset.filter(tenant=tenant)
    return queryset

3. Forgot to Set on_delete for ForeignKey

Wrong:

tenant = models.ForeignKey(Tenant)  # Missing on_delete!

Correct:

tenant = models.ForeignKey(Tenant, on_delete=models.CASCADE)

4. Used save() Without Hashing Password

Wrong:

user = User(**validated_data)
user.save()  # Password stored in plaintext!

Correct:

password = validated_data.pop('password')
user = User(**validated_data)
user.set_password(password)  # Hash password
user.save()

5. Middleware Order Incorrect

Wrong:

MIDDLEWARE = [
    'apps.tenants.middleware.TenantMiddleware',  # TOO EARLY!
    'django.contrib.auth.middleware.AuthenticationMiddleware',
]

Correct:

MIDDLEWARE = [
    'django.contrib.auth.middleware.AuthenticationMiddleware',
    'apps.tenants.middleware.TenantMiddleware',  # AFTER auth
]

Quick Reference Tables

HTTP Methods → ViewSet Methods

HTTP	FastAPI	Django ViewSet	URL Pattern
GET	`@app.get("/users")`	`def list(self, request)`	`/users/`
POST	`@app.post("/users")`	`def create(self, request)`	`/users/`
GET	`@app.get("/users/{id}")`	`def retrieve(self, request, pk)`	`/users/{pk}/`
PUT	`@app.put("/users/{id}")`	`def update(self, request, pk)`	`/users/{pk}/`
PATCH	`@app.patch("/users/{id}")`	`def partial_update(self, request, pk)`	`/users/{pk}/`
DELETE	`@app.delete("/users/{id}")`	`def destroy(self, request, pk)`	`/users/{pk}/`

SQLAlchemy → Django ORM

SQLAlchemy	Django ORM	Notes
`select(User)`	`User.objects.all()`	SELECT all
`.filter(User.id == id)`	`.filter(id=id)`	WHERE clause
`.offset(skip).limit(limit)`	`[skip:skip+limit]`	Pagination
`db.add(user)`	`user.save()`	INSERT
`db.delete(user)`	`user.delete()`	DELETE
`await db.commit()`	Automatic on save()	Commit
`selectinload(relationship)`	`prefetch_related('relationship')`	Eager loading

Pydantic → DRF Serializers

Pydantic	DRF	Notes
`class Schema(BaseModel)`	`class Serializer(serializers.ModelSerializer)`	Base class
`Field(..., max_length=255)`	`class Meta: model = User`	Auto from model
`@validator('field')`	`def validate_field(self, value)`	Field validation
`@root_validator`	`def validate(self, attrs)`	Object validation
`class Config: from_attributes = True`	Not needed	Auto-handled

Summary

Key Conversion Points:

Models: SQLAlchemy → Django ORM with TenantModel base class
Serializers: Pydantic → Django REST Framework serializers
Views: FastAPI route handlers → DRF ViewSets
Auth: Dependency injection → Authentication/Permission classes
Database: Async sessions → Django ORM (sync by default)
Validation: Pydantic validators → Serializer validation methods
Middleware: Starlette → Django middleware
Tenant Context: Manual filtering → Automatic RLS + middleware

Advantages of Django Approach:

✅ Row-Level Security (RLS) - Database-enforced tenant isolation
✅ Django Admin - Free admin interface
✅ Mature ecosystem - More packages, better documentation
✅ Simpler ORM - Less boilerplate than SQLAlchemy
✅ Built-in auth - User model, permissions, password hashing
✅ Automatic migrations - makemigrations and migrate

References:

ADR-007: Django Multi-Tenant Architecture (this document's source)
Django REST Framework: https://www.django-rest-framework.org/
django-multitenant: https://github.com/citusdata/django-multitenant
PostgreSQL RLS: https://www.postgresql.org/docs/15/ddl-rowsecurity.html

Last Updated: 2025-11-30 Status: Active - Use for FastAPI → Django conversion Related: adr-007-django-multitenant-architecture.md

Table of Contents​

1. Import Statements​

FastAPI Pattern​

Django Equivalent​

Key Differences​

2. Request Handling​

FastAPI Pattern​

Django Equivalent​

Key Differences​

3. Model Definitions​

FastAPI Pattern (SQLAlchemy)​

Django Equivalent (From ADR-007)​

Key Differences​

Tenant-Scoped Model (From ADR-007)​

4. Serializers (Pydantic → DRF)​

FastAPI Pattern (Pydantic)​

Django Equivalent (From ADR-007 Pattern)​

Key Differences​

5. ViewSets (Route Handlers → ViewSets)​

FastAPI Pattern​

Django Equivalent (From ADR-007)​

URL Configuration (NEW)​

Key Differences​

6. Middleware​

FastAPI Pattern​

Django Equivalent (From ADR-007)​

Tenant Context Module (From ADR-007)​

Settings Configuration (From ADR-007)​

Key Differences​

7. Authentication​

FastAPI Pattern​

Django Equivalent​

Permission Classes​

Usage in ViewSets​

Settings Configuration​

Key Differences​

8. Validation​

FastAPI Pattern (Pydantic)​

Django Equivalent​

Custom Validators (Reusable)​

Model-Level Validation​

Key Differences​

9. Async Handling​

FastAPI Pattern (Native Async)​

Django Equivalent (Django 4.2+ Async Views)​

Async ORM (Django 4.1+)​

Background Tasks (Celery)​

Settings Configuration​

Key Differences​

10. Database Access​

FastAPI Pattern (SQLAlchemy Async)​

Django Equivalent (Django ORM)​

Tenant Context in Queries (From ADR-007)​

Key Differences​

11. Complete File Examples​

Example 1: User Management (Complete Conversion)​

FastAPI Version (Before)​

Django Version (After)​

Example 2: License Session (With Redis Integration)​

Django Version (Full Implementation)​

Conversion Checklist​

For Each Model File​

For Each Serializer File​

For Each View File​

For URLs​

For Settings​

For Testing​

Migration Path​

Common Pitfalls​

1. Forgot to Add Tenant to Model​

2. Manual Tenant Filtering Instead of Using Context​

3. Forgot to Set on_delete for ForeignKey​

4. Used save() Without Hashing Password​

5. Middleware Order Incorrect​

Quick Reference Tables​

HTTP Methods → ViewSet Methods​

SQLAlchemy → Django ORM​

Pydantic → DRF Serializers​

Summary​

Table of Contents

1. Import Statements

FastAPI Pattern

Django Equivalent

Key Differences

2. Request Handling

FastAPI Pattern

Django Equivalent

Key Differences

3. Model Definitions

FastAPI Pattern (SQLAlchemy)

Django Equivalent (From ADR-007)

Key Differences

Tenant-Scoped Model (From ADR-007)

4. Serializers (Pydantic → DRF)

FastAPI Pattern (Pydantic)

Django Equivalent (From ADR-007 Pattern)

Key Differences

5. ViewSets (Route Handlers → ViewSets)

FastAPI Pattern

Django Equivalent (From ADR-007)

URL Configuration (NEW)

Key Differences

6. Middleware

FastAPI Pattern

Django Equivalent (From ADR-007)

Tenant Context Module (From ADR-007)

Settings Configuration (From ADR-007)

Key Differences

7. Authentication

FastAPI Pattern

Django Equivalent

Permission Classes

Usage in ViewSets

Settings Configuration

Key Differences

8. Validation

FastAPI Pattern (Pydantic)

Django Equivalent

Custom Validators (Reusable)

Model-Level Validation

Key Differences

9. Async Handling

FastAPI Pattern (Native Async)

Django Equivalent (Django 4.2+ Async Views)

Async ORM (Django 4.1+)

Background Tasks (Celery)

Settings Configuration

Key Differences

10. Database Access

FastAPI Pattern (SQLAlchemy Async)

Django Equivalent (Django ORM)

Tenant Context in Queries (From ADR-007)

Key Differences

11. Complete File Examples

Example 1: User Management (Complete Conversion)

FastAPI Version (Before)

Django Version (After)

Example 2: License Session (With Redis Integration)

Django Version (Full Implementation)

Conversion Checklist

For Each Model File

For Each Serializer File

For Each View File

For URLs

For Settings

For Testing

Migration Path

Common Pitfalls

1. Forgot to Add Tenant to Model

2. Manual Tenant Filtering Instead of Using Context

3. Forgot to Set on_delete for ForeignKey

4. Used save() Without Hashing Password

5. Middleware Order Incorrect

Quick Reference Tables

HTTP Methods → ViewSet Methods

SQLAlchemy → Django ORM

Pydantic → DRF Serializers

Summary