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
Layered Architecture
-
Routes: HTTP handling, validation
-
Services: Business logic
-
Repositories: Data access
Async Everything
-
Use async def for all endpoints
-
Use async database drivers
-
Avoid blocking calls in async context
Type Safety
-
Use Pydantic for all I/O
-
Use SQLAlchemy 2.0 Mapped types
-
Enable strict mypy
Dependency Injection
-
Use FastAPI Depends() throughout
-
Inject sessions, services, settings
Error Handling
-
Use HTTPException for API errors
-
Implement exception handlers for common cases
-
Return consistent error responses
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