fastapi

name	fastapi
description	Python FastAPI expert for async APIs, SQLAlchemy 2.0, Pydantic v2, uv package manager, and production patterns. Use PROACTIVELY for Python API projects, async debugging, and database integration.

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 () -> User:
        user = User(**data)
        .db.add(user)
         .db.flush()
         .db.refresh(user)
         user

      () -> User:
         field, value  data.items():
             value   :
                (user, field, value)
         .db.flush()
         .db.refresh(user)
         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 () -> User:
        data = user_in.model_dump()
        data[] = hash_password(data.pop())
          .repo.create(data)

      () -> User:
        data = user_in.model_dump(exclude_unset=)
           data:
            data[] = hash_password(data.pop())
          .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")
     user



  ():
      service.get_by_email(user_in.email):
         HTTPException(status_code=, detail=)
      service.create(user_in)



  ():
    user =  service.get(user_id)
      user:
         HTTPException(status_code=, detail=)
      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