Django to FastAPI Migration Guide

---

When to Use This Guide

Migrate to FastAPI when:

• Need async/await for I/O-bound operations
• Require WebSocket or Server-Sent Events
• Want automatic OpenAPI/Swagger documentation
• Need better performance for API-heavy workloads
• Desire modern Python type hints and editor support
• Building microservices from Django monolith
• Require lower resource consumption

DO NOT migrate when:

• Heavy use of Django admin interface
• Extensive Django ORM model inheritance
• Complex form handling and validation
• Server-side template rendering required
• Team lacks async Python experience
• Django ecosystem plugins are critical
• Migration cost exceeds business value

---

Concept Mapping: Django/DRF → FastAPI

---

Serializer → Pydantic V2 Migration

Django REST Framework Serializer

# Django DRF
from rest_framework import serializers
from .models import User, Post

class UserSerializer(serializers.ModelSerializer):
    post_count = serializers.SerializerMethodField()

    class Meta:
        model = User
        fields = ['id', 'username', 'email', 'created_at', 'post_count']
        read_only_fields = ['id', 'created_at']
        extra_kwargs = {
            'email': {'write_only': True}
        }

    def get_post_count(self, obj):
        return obj.posts.count()

    def validate_username(self, value):
        if len(value) < 3:
            raise serializers.ValidationError("Username too short")
        return value

class PostSerializer(serializers.ModelSerializer):
    author = UserSerializer(read_only=True)
    tags = serializers.ListField(child=serializers.CharField())

    class Meta:
        model = Post
        fields = ['id', 'title', 'content', 'author', 'tags', 'published']

    def create(self, validated_data):
        tags = validated_data.pop('tags', [])
        post = Post.objects.create(**validated_data)
        post.tags.set(tags)
        return post

FastAPI Pydantic V2 Schemas

# FastAPI with Pydantic V2
from pydantic import BaseModel, EmailStr, Field, field_validator, computed_field
from datetime import datetime
from typing import Annotated

# Base schemas
class UserBase(BaseModel):
    username: Annotated[str, Field(min_length=3, max_length=50)]
    email: EmailStr

# Create schema (input)
class UserCreate(UserBase):
    password: Annotated[str, Field(min_length=8)]

    @field_validator('username')
    @classmethod
    def validate_username(cls, v: str) -> str:
        if len(v) < 3:
            raise ValueError("Username too short")
        return v

# Update schema (partial)
class UserUpdate(BaseModel):
    username: Annotated[str | None, Field(min_length=3, max_length=50)] = None
    email: EmailStr | None = None

# Read schema (output) - analogous to read_only_fields
class UserRead(UserBase):
    id: int
    created_at: datetime

    model_config = {
        "from_attributes": True  # Pydantic V2: replaces orm_mode
    }

# Read schema with relations - analogous to SerializerMethodField
class UserReadWithStats(UserRead):
    post_count: int

    @computed_field  # Pydantic V2 computed fields
    @property
    def display_name(self) -> str:
        return f"@{self.username}"

# Nested schemas
class PostBase(BaseModel):
    title: Annotated[str, Field(max_length=200)]
    content: str
    tags: list[str] = []
    published: bool = False

class PostCreate(PostBase):
    pass

class PostRead(PostBase):
    id: int
    author: UserRead  # Nested serialization
    created_at: datetime

    model_config = {"from_attributes": True}

# Embedding vs side-loading
class PostReadMinimal(BaseModel):
    """Minimal post representation (just ID)"""
    id: int
    title: str
    author_id: int  # Side-loaded reference

    model_config = {"from_attributes": True}

---

ViewSet → APIRouter Migration

Django REST Framework ViewSet

# Django DRF ViewSet
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 django.shortcuts import get_object_or_404

class PostViewSet(viewsets.ModelViewSet):
    queryset = Post.objects.all()
    serializer_class = PostSerializer
    permission_classes = [IsAuthenticated]

    def get_queryset(self):
        queryset = super().get_queryset()
        if self.request.user.is_authenticated:
            return queryset.filter(author=self.request.user)
        return queryset.none()

    def perform_create(self, serializer):
        serializer.save(author=self.request.user)

    @action(detail=True, methods=['post'])
    def publish(self, request, pk=None):
        post = self.get_object()
        post.published = True
        post.save()
        return Response({'status': 'published'})

    @action(detail=False, methods=['get'])
    def recent(self, request):
        recent_posts = self.get_queryset().order_by('-created_at')[:10]
        serializer = self.get_serializer(recent_posts, many=True)
        return Response(serializer.data)

FastAPI APIRouter with Dependencies

# FastAPI APIRouter
from fastapi import APIRouter, Depends, HTTPException, status, Query
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from typing import Annotated

from .database import get_db
from .auth import get_current_user
from .models import Post as PostModel, User as UserModel
from .schemas import PostRead, PostCreate, PostUpdate, UserRead

router = APIRouter(prefix="/posts", tags=["posts"])

# Dependency for database session
DbSession = Annotated[AsyncSession, Depends(get_db)]
CurrentUser = Annotated[UserModel, Depends(get_current_user)]

# List posts (GET /posts)
@router.get("/", response_model=list[PostRead])
async def list_posts(
    db: DbSession,
    current_user: CurrentUser,
    skip: int = Query(0, ge=0),
    limit: int = Query(100, le=100),
):
    """Analogous to ViewSet.list()"""
    result = await db.execute(
        select(PostModel)
        .where(PostModel.author_id == current_user.id)
        .offset(skip)
        .limit(limit)
    )
    posts = result.scalars().all()
    return posts

# Create post (POST /posts)
@router.post("/", response_model=PostRead, status_code=status.HTTP_201_CREATED)
async def create_post(
    post_data: PostCreate,
    db: DbSession,
    current_user: CurrentUser,
):
    """Analogous to ViewSet.create()"""
    post = PostModel(**post_data.model_dump(), author_id=current_user.id)
    db.add(post)
    await db.commit()
    await db.refresh(post)
    return post

# Retrieve single post (GET /posts/{post_id})
@router.get("/{post_id}", response_model=PostRead)
async def get_post(
    post_id: int,
    db: DbSession,
    current_user: CurrentUser,
):
    """Analogous to ViewSet.retrieve()"""
    result = await db.execute(
        select(PostModel).where(
            PostModel.id == post_id,
            PostModel.author_id == current_user.id
        )
    )
    post = result.scalar_one_or_none()
    if not post:
        raise HTTPException(status_code=404, detail="Post not found")
    return post

# Update post (PUT /posts/{post_id})
@router.put("/{post_id}", response_model=PostRead)
async def update_post(
    post_id: int,
    post_data: PostUpdate,
    db: DbSession,
    current_user: CurrentUser,
):
    """Analogous to ViewSet.update()"""
    result = await db.execute(
        select(PostModel).where(
            PostModel.id == post_id,
            PostModel.author_id == current_user.id
        )
    )
    post = result.scalar_one_or_none()
    if not post:
        raise HTTPException(status_code=404, detail="Post not found")

    # Update only provided fields
    for field, value in post_data.model_dump(exclude_unset=True).items():
        setattr(post, field, value)

    await db.commit()
    await db.refresh(post)
    return post

# Delete post (DELETE /posts/{post_id})
@router.delete("/{post_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_post(
    post_id: int,
    db: DbSession,
    current_user: CurrentUser,
):
    """Analogous to ViewSet.destroy()"""
    result = await db.execute(
        select(PostModel).where(
            PostModel.id == post_id,
            PostModel.author_id == current_user.id
        )
    )
    post = result.scalar_one_or_none()
    if not post:
        raise HTTPException(status_code=404, detail="Post not found")

    await db.delete(post)
    await db.commit()

# Custom action: Publish (POST /posts/{post_id}/publish)
@router.post("/{post_id}/publish", response_model=dict)
async def publish_post(
    post_id: int,
    db: DbSession,
    current_user: CurrentUser,
):
    """Analogous to @action(detail=True)"""
    result = await db.execute(
        select(PostModel).where(
            PostModel.id == post_id,
            PostModel.author_id == current_user.id
        )
    )
    post = result.scalar_one_or_none()
    if not post:
        raise HTTPException(status_code=404, detail="Post not found")

    post.published = True
    await db.commit()
    return {"status": "published"}

# Custom collection action: Recent posts (GET /posts/recent)
@router.get("/actions/recent", response_model=list[PostRead])
async def recent_posts(
    db: DbSession,
    current_user: CurrentUser,
    limit: int = Query(10, le=50),
):
    """Analogous to @action(detail=False)"""
    result = await db.execute(
        select(PostModel)
        .where(PostModel.author_id == current_user.id)
        .order_by(PostModel.created_at.desc())
        .limit(limit)
    )
    posts = result.scalars().all()
    return posts

---

Django ORM → Async SQLAlchemy

Django ORM Models

# Django models
from django.db import models

class User(models.Model):
    username = models.CharField(max_length=50, unique=True)
    email = models.EmailField(unique=True)
    created_at = models.DateTimeField(auto_now_add=True)

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

class Post(models.Model):
    title = models.CharField(max_length=200)
    content = models.TextField()
    author = models.ForeignKey(User, on_delete=models.CASCADE, related_name='posts')
    created_at = models.DateTimeField(auto_now_add=True)
    published = models.BooleanField(default=False)

    class Meta:
        db_table = 'posts'
        ordering = ['-created_at']

SQLAlchemy 2.0 Async Models

# SQLAlchemy 2.0 models
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
from sqlalchemy import String, Text, Boolean, ForeignKey, Index
from datetime import datetime
from typing import List

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = 'users'

    # Primary key
    id: Mapped[int] = mapped_column(primary_key=True)

    # Columns with type hints
    username: Mapped[str] = mapped_column(String(50), unique=True, index=True)
    email: Mapped[str] = mapped_column(String(255), unique=True)
    created_at: Mapped[datetime] = mapped_column(default=datetime.utcnow)

    # Relationships (analogous to related_name)
    posts: Mapped[List["Post"]] = relationship(back_populates="author")

    __table_args__ = (
        Index('ix_users_username', 'username'),
    )

class Post(Base):
    __tablename__ = 'posts'

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))
    content: Mapped[str] = mapped_column(Text)
    author_id: Mapped[int] = mapped_column(ForeignKey('users.id', ondelete='CASCADE'))
    created_at: Mapped[datetime] = mapped_column(default=datetime.utcnow)
    published: Mapped[bool] = mapped_column(Boolean, default=False)

    # Relationship
    author: Mapped["User"] = relationship(back_populates="posts")

    __table_args__ = (
        Index('ix_posts_created_at', 'created_at'),
    )

Query Patterns: Django ORM vs SQLAlchemy

# Django ORM queries
from django.db.models import Count, Q

# Simple filter
posts = Post.objects.filter(published=True)

# Select related (JOIN)
posts = Post.objects.select_related('author').filter(published=True)

# Prefetch related (separate query)
users = User.objects.prefetch_related('posts').all()

# Complex filtering
posts = Post.objects.filter(
    Q(published=True) | Q(author__username='admin')
).order_by('-created_at')[:10]

# Aggregation
user_stats = User.objects.annotate(
    post_count=Count('posts')
).filter(post_count__gte=5)

# SQLAlchemy 2.0 async queries
from sqlalchemy import select, func, or_
from sqlalchemy.orm import selectinload, joinedload

# Simple filter
async def get_published_posts(db: AsyncSession):
    result = await db.execute(
        select(Post).where(Post.published == True)
    )
    return result.scalars().all()

# Eager loading with JOIN (selectinload = separate query)
async def get_posts_with_authors(db: AsyncSession):
    result = await db.execute(
        select(Post)
        .options(selectinload(Post.author))
        .where(Post.published == True)
    )
    return result.scalars().all()

# Prefetch related (joinedload = single query with JOIN)
async def get_users_with_posts(db: AsyncSession):
    result = await db.execute(
        select(User).options(joinedload(User.posts))
    )
    return result.unique().scalars().all()

# Complex filtering
async def get_complex_posts(db: AsyncSession):
    result = await db.execute(
        select(Post)
        .join(Post.author)
        .where(
            or_(
                Post.published == True,
                User.username == 'admin'
            )
        )
        .order_by(Post.created_