FastAPI Development Expert

Expert in Python FastAPI development with async patterns, SQLAlchemy 2.0, Pydantic v2, and modern Python tooling. Specialized in building scalable, production-ready APIs.

When to Use

• Python FastAPI projects
• Async API development
• SQLAlchemy database integration
• Pydantic schema design
• Python backend debugging

Technology Stack

Core

• Python 3.12+: Modern Python with type hints
• FastAPI: Async web framework
• Pydantic v2: Data validation and settings
• uv: Fast package manager (preferred over pip/poetry)

Database

• SQLAlchemy 2.0: Async ORM with mapped columns
• Alembic: Database migrations
• asyncpg: PostgreSQL async driver
• aiosqlite: SQLite async driver (testing)

Testing & Quality

• pytest: Testing framework
• pytest-asyncio: Async test support
• httpx: Async HTTP client for testing
• ruff: Fast linter and formatter
• mypy: Static type checking

Project Structure

Feature-based modular architecture - code organized by domain, not by layer:

project/
├── pyproject.toml
├── uv.lock
├── .python-version
├── .env
├── alembic/
│   ├── env.py
│   └── versions/
├── src/
│   └── app/
│       ├── __init__.py
│       ├── main.py              # FastAPI app entry
│       ├── config.py            # Settings
│       ├── database.py          # DB session
│       ├── core/
│       │   ├── __init__.py
│       │   ├── dependencies.py  # Shared dependencies
│       │   ├── exceptions.py    # Custom exceptions
│       │   ├── middleware.py    # Middleware
│       │   └── security.py      # Auth utilities
│       ├── models/
│       │   ├── __init__.py
│       │   └── base.py          # SQLAlchemy base & mixins
│       ├── features/
│       │   ├── __init__.py
│       │   ├── auth/
│       │   │   ├── __init__.py
│       │   │   ├── api.py       # Auth endpoints
│       │   │   ├── schemas.py   # Auth Pydantic schemas
│       │   │   ├── services.py  # Auth business logic
│       │   │   ├── models.py    # Auth SQLAlchemy models
│       │   │   └── utils.py     # Auth helpers
│       │   ├── users/
│       │   │   ├── __init__.py
│       │   │   ├── api.py       # User endpoints
│       │   │   ├── schemas.py   # User Pydantic schemas
│       │   │   ├── services.py  # User business logic
│       │   │   ├── models.py    # User SQLAlchemy models
│       │   │   └── repository.py # User data access
│       │   └── items/
│       │       ├── __init__.py
│       │       ├── api.py
│       │       ├── schemas.py
│       │       ├── services.py
│       │       └── models.py
│       └── api/
│           ├── __init__.py
│           └── router.py        # Aggregates all feature routers
└── tests/
    ├── conftest.py
    ├── features/
    │   ├── auth/
    │   │   └── test_auth.py
    │   └── users/
    │       └── test_users.py
    └── integration/

Code Patterns

Configuration with pydantic-settings

from functools import lru_cache
from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_file=".env",
        case_sensitive=False,
    )

    app_name: str = "My API"
    debug: bool = False
    database_url: str = "postgresql+asyncpg://user:pass@localhost/db"
    secret_key: str = "change-me"


@lru_cache
def get_settings() -> Settings:
    return Settings()

Async Database Session

from collections.abc import AsyncGenerator
from sqlalchemy.ext.asyncio import (
    AsyncSession,
    async_sessionmaker,
    create_async_engine,
)

engine = create_async_engine(settings.database_url, echo=settings.debug)
async_session_maker = async_sessionmaker(engine, expire_on_commit=False)


async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with async_session_maker() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

SQLAlchemy 2.0 Model

from datetime import datetime
from sqlalchemy import String, DateTime, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(255), unique=True, index=True)
    full_name: Mapped[str | None] = mapped_column(String(255))
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), server_default=func.now()
    )

Pydantic Schemas

from pydantic import BaseModel, EmailStr, ConfigDict


class UserBase(BaseModel):
    email: EmailStr
    full_name: str | None = None


class UserCreate(UserBase):
    password: str


class UserResponse(UserBase):
    model_config = ConfigDict(from_attributes=True)
    id: int

Feature Module Pattern

Each feature is self-contained with its own api, schemas, services, models, and utils.

Feature: users/schemas.py

from pydantic import BaseModel, EmailStr, ConfigDict


class UserBase(BaseModel):
    email: EmailStr
    full_name: str | None = None


class UserCreate(UserBase):
    password: str


class UserUpdate(BaseModel):
    email: EmailStr | None = None
    full_name: str | None = None
    password: str | None = None


class UserResponse(UserBase):
    model_config = ConfigDict(from_attributes=True)
    id: int
    is_active: bool

Feature: users/models.py

from sqlalchemy import String
from sqlalchemy.orm import Mapped, mapped_column

from app.models.base import Base, TimestampMixin


class User(Base, TimestampMixin):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(255), unique=True, index=True)
    hashed_password: Mapped[str] = mapped_column(String(255))
    full_name: Mapped[str | None] = mapped_column(String(255))
    is_active: Mapped[bool] = mapped_column(default=True)

Feature: users/repository.py

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.features.users.models import User


class UserRepository:
    def __init__(self, db: AsyncSession):
        self.db = db

    async def get(self, id: int) -> User | None:
        result = await self.db.execute(select(User).where(User.id == id))
        return result.scalar_one_or_none()

    async def get_by_email(self, email: str) -> User | None:
        result = await self.db.execute(select(User).where(User.email == email))
        return result.scalar_one_or_none()

    async def get_all(self, skip: int = 0, limit: int = 100) -> list[User]:
        result = await self.db.execute(select(User).offset(skip).limit(limit))
        return list(result.scalars().all())

    async def create(self, data: dict) -> User:
        user = User(**data)
        self.db.add(user)
        await self.db.flush()
        await self.db.refresh(user)
        return user

    async def update(self, user: User, data: dict) -> User:
        for field, value in data.items():
            if value is not None:
                setattr(user, field, value)
        await self.db.flush()
        await self.db.refresh(user)
        return user

Feature: users/services.py

from sqlalchemy.ext.asyncio import AsyncSession

from app.core.security import hash_password
from app.features.users.models import User
from app.features.users.repository import UserRepository
from app.features.users.schemas import UserCreate, UserUpdate


class UserService:
    def __init__(self, db: AsyncSession):
        self.db = db
        self.repo = UserRepository(db)

    async def get(self, user_id: int) -> User | None:
        return await self.repo.get(user_id)

    async def get_by_email(self, email: str) -> User | None:
        return await self.repo.get_by_email(email)

    async def list(self, skip: int = 0, limit: int = 100) -> list[User]:
        return await self.repo.get_all(skip=skip, limit=limit)

    async def create(self, user_in: UserCreate) -> User:
        data = user_in.model_dump()
        data["hashed_password"] = hash_password(data.pop("password"))
        return await self.repo.create(data)

    async def update(self, user: User, user_in: UserUpdate) -> User:
        data = user_in.model_dump(exclude_unset=True)
        if "password" in data:
            data["hashed_password"] = hash_password(data.pop("password"))
        return await self.repo.update(user, data)

Feature: users/api.py

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

from app.database import get_db
from app.features.users.schemas import UserCreate, UserResponse, UserUpdate
from app.features.users.services import UserService

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


def get_service(db: AsyncSession = Depends(get_db)) -> UserService:
    return UserService(db)


@router.get("", response_model=list[UserResponse])
async def list_users(
    skip: int = 0,
    limit: int = 100,
    service: UserService = Depends(get_service),
):
    return await service.list(skip=skip, limit=limit)


@router.get("/{user_id}", response_model=UserResponse)
async def get_user(
    user_id: int,
    service: UserService = Depends(get_service),
):
    user = await service.get(user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user


@router.post("", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
async def create_user(
    user_in: UserCreate,
    service: UserService = Depends(get_service),
):
    if await service.get_by_email(user_in.email):
        raise HTTPException(status_code=400, detail="Email already registered")
    return await service.create(user_in)


@router.patch("/{user_id}", response_model=UserResponse)
async def update_user(
    user_id: int,
    user_in: UserUpdate,
    service: UserService = Depends(get_service),
):
    user = await service.get(user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return await service.update(user, user_in)

Feature: users/init.py (exports)

from app.features.users.api import router
from app.features.users.models import User
from app.features.users.schemas import UserCreate, UserResponse, UserUpdate
from app.features.users.services import UserService

__all__ = ["router", "User", "UserCreate", "UserResponse", "UserUpdate", "UserService"]

Main Router (app/api/router.py)

from fastapi import APIRouter

from app.features.auth import router as auth_router
from app.features.users import router as users_router
from app.features.items import router as items_router

api_router = APIRouter()
api_router.include_router(auth_router)
api_router.include_router(users_router)
api_router.include_router(items_router)

App with Lifespan

from contextlib import asynccontextmanager
from collections.abc import AsyncIterator
from fastapi import FastAPI


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    # Startup
    yield
    # Shutdown
    await engine.dispose()


app = FastAPI(title=settings.app_name, lifespan=lifespan)
app.include_router(api_router, prefix="/api/v1")

Common Commands

# Project setup with uv
uv init my-project && cd my-project
uv python pin 3.12
uv add fastapi uvicorn[standard] sqlalchemy[asyncio] asyncpg pydantic-settings
uv add --dev pytest pytest-asyncio httpx ruff mypy

# Run development server
uv run uvicorn app.main:app --reload

# Database migrations
uv run alembic revision --autogenerate -m "Add users table"
uv run alembic upgrade head

# Testing
uv run pytest -v
uv run pytest --cov=app

# Code quality
uv run ruff check . --fix
uv run ruff format .
uv run mypy src/

Best Practices

1. Layered Architecture

   • Routes: HTTP handling, validation
   • Services: Business logic
   • Repositories: Data access

2. Async Everything

   • Use async def for all endpoints
   • Use async database drivers
   • Avoid blocking calls in async context

3. Type Safety

   • Use Pydantic for all I/O
   • Use SQLAlchemy 2.0 Mapped types
   • Enable strict mypy

4. Dependency Injection

   • Use FastAPI Depends() throughout
   • Inject sessions, services, settings

5. Error Handling

   • Use HTTPException for API errors
   • Implement exception handlers for common cases
   • Return consistent error responses

6. Testing

   • Use in-memory SQLite for unit tests
   • Override dependencies in tests
   • Test services independently from routes

Common Issues

Async Session Not Committing

• Ensure await session.commit() is called
• Check that exceptions trigger rollback
• Use context manager pattern

Circular Imports

• Import models in alembic/env.py
• Use TYPE_CHECKING for type hints
• Structure imports carefully

N+1 Query Problem

• Use selectinload() or joinedload()
• Review queries with echo=True
• Profile with database logs

Pydantic v2 Migration

• Use model_dump() not dict()
• Use ConfigDict not class Config
• Use from_attributes=True for ORM mode