fastapi-workflow

name	fastapi-workflow
description	FastAPI framework workflow guidelines. Activate when working with FastAPI projects, uvicorn, or FastAPI-specific patterns.
location	user

The key words "MUST", "MUST NOT", "REQUIRED", "SHALL", "SHALL NOT", "SHOULD", "SHOULD NOT", "RECOMMENDED", "MAY", and "OPTIONAL" in this document are to be interpreted as described in RFC 2119.

FastAPI Workflow

Tool Grid

Task	Tool	Command
Run dev	Uvicorn	uvicorn app.main:app --reload
Test	pytest + httpx	uv run pytest
Docs	Built-in	/docs or /redoc
Lint	Ruff	uv run ruff check .
Format	Ruff	uv run ruff format .
Type check	mypy	uv run mypy .

---

Project Structure

project/
├── app/
│   ├── __init__.py
│   ├── main.py              # FastAPI app instance
│   ├── config.py            # BaseSettings configuration
│   ├── dependencies.py      # Shared Depends() callables
│   ├── exceptions.py        # Custom exception handlers
│   ├── middleware.py        # Custom middleware
│   ├── models/              # SQLAlchemy/Pydantic models
│   │   ├── __init__.py
│   │   ├── domain.py        # SQLAlchemy ORM models
│   │   └── schemas.py       # Pydantic schemas
│   ├── routers/             # APIRouter modules
│   │   ├── __init__.py
│   │   ├── users.py
│   │   └── items.py
│   ├── services/            # Business logic layer
│   │   └── __init__.py
│   └── db/                  # Database configuration
│       ├── __init__.py
│       └── session.py
├── tests/
│   ├── conftest.py          # Fixtures
│   └── test_*.py
├── pyproject.toml
└── .env

---

Dependency Injection

Dependencies MUST use Depends() for:

• Database sessions
• Authentication/authorization
• Configuration access
• Shared services

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

from app.db.session import get_db
from app.config import Settings, get_settings

router = APIRouter()

# Database session dependency
async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with async_session_maker() as session:
        yield session

# Settings dependency
def get_settings() -> Settings:
    return Settings()

# Service with injected dependencies
class UserService:
    def __init__(
        self,
        db: AsyncSession = Depends(get_db),
        settings: Settings = Depends(get_settings),
    ):
        self.db = db
        self.settings = settings

# Route using dependency
@router.get("/users/{user_id}")
async def get_user(
    user_id: int,
    service: UserService = Depends(),
) -> UserResponse:
    return await service.get_user(user_id)

Dependency Caching

• Dependencies are cached per-request by default
• Use Depends(get_db, use_cache=False) to disable caching when needed

---

Pydantic Validation

Request and response models MUST use Pydantic BaseModel or dataclasses.

from pydantic import BaseModel, Field, EmailStr, ConfigDict
from datetime import datetime

# Request schema
class UserCreate(BaseModel):
    email: EmailStr
    name: str = Field(..., min_length=1, max_length=100)
    age: int = Field(..., ge=0, le=150)

# Response schema with ORM mode
class UserResponse(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    id: int
    email: EmailStr
    name: str
    created_at: datetime

# Nested models
class UserWithItems(UserResponse):
    items: list[ItemResponse] = []

Validation Rules

• MUST define explicit Field constraints for all user inputs
• MUST use from_attributes=True for ORM model conversion
• SHOULD use EmailStr, HttpUrl, and other specialized types
• MUST NOT expose internal fields in response models

---

Configuration with BaseSettings

All configuration MUST use Pydantic BaseSettings.

from pydantic_settings import BaseSettings, SettingsConfigDict
from functools import lru_cache

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

    # Database
    database_url: str
    database_pool_size: int = 5

    # API
    api_prefix: str = "/api/v1"
    debug: bool = False

    # Security
    secret_key: str
    access_token_expire_minutes: int = 30

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

Configuration Rules

• MUST NOT hardcode secrets or environment-specific values
• MUST use .env files for local development
• SHOULD use @lru_cache for settings singleton
• Environment variables MUST override .env values

---

Async Database Access

Database operations MUST be async using SQLAlchemy 2.0+ async.

from sqlalchemy.ext.asyncio import (
    AsyncSession,
    async_sessionmaker,
    create_async_engine,
)
from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
    pass

# Engine setup
engine = create_async_engine(
    settings.database_url,
    echo=settings.debug,
    pool_size=settings.database_pool_size,
)

async_session_maker = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,
)

# Dependency
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

Database Rules

• MUST use async drivers (asyncpg, aiosqlite)
• MUST handle session lifecycle in dependencies
• SHOULD use expire_on_commit=False for async sessions
• MUST NOT use synchronous database calls

---

Router Organization

Routers MUST be organized by domain with clear prefixes and tags.

# app/routers/users.py
from fastapi import APIRouter, Depends, status

router = APIRouter(
    prefix="/users",
    tags=["users"],
    responses={404: {"description": "Not found"}},
)

@router.post("/", status_code=status.HTTP_201_CREATED)
async def create_user(user: UserCreate) -> UserResponse:
    ...

@router.get("/{user_id}")
async def get_user(user_id: int) -> UserResponse:
    ...

# app/main.py
from fastapi import FastAPI
from app.routers import users, items
from app.config import get_settings

settings = get_settings()

app = FastAPI(
    title="My API",
    version="1.0.0",
    openapi_url=f"{settings.api_prefix}/openapi.json",
)

app.include_router(users.router, prefix=settings.api_prefix)
app.include_router(items.router, prefix=settings.api_prefix)

Router Rules

• MUST use descriptive tags for OpenAPI grouping
• MUST define common responses at router level
• SHOULD use status codes from fastapi.status
• Routers SHOULD NOT contain business logic

---

Exception Handling

from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse

# Custom exception
class NotFoundError(Exception):
    def __init__(self, resource: str, id: int):
        self.resource = resource
        self.id = id

# Exception handler
@app.exception_handler(NotFoundError)
async def not_found_handler(request: Request, exc: NotFoundError) -> JSONResponse:
    return JSONResponse(
        status_code=404,
        content={"detail": f"{exc.resource} with id {exc.id} not found"},
    )

# Usage in route
@router.get("/{user_id}")
async def get_user(user_id: int, db: AsyncSession = Depends(get_db)) -> UserResponse:
    user = await db.get(User, user_id)
    if not user:
        raise NotFoundError("User", user_id)
    return user

Exception Rules

• MUST use HTTPException for standard HTTP errors
• SHOULD define custom exceptions for domain errors
• MUST register exception handlers on app instance
• MUST NOT expose internal errors to clients

---

Testing with TestClient

import pytest
from httpx import AsyncClient, ASGITransport
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker

from app.main import app
from app.db.session import get_db
from app.config import get_settings, Settings

# Override settings
def get_settings_override() -> Settings:
    return Settings(database_url="sqlite+aiosqlite:///:memory:")

app.dependency_overrides[get_settings] = get_settings_override

@pytest.fixture
async def client():
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test",
    ) as ac:
        yield ac

@pytest.mark.asyncio
async def test_create_user(client: AsyncClient):
    response = await client.post(
        "/api/v1/users/",
        json={"email": "[e-mail]", "name": "Test", "age": 25},
    )
    assert response.status_code == 201
    assert response.json()["email"] == "[e-mail]"

Testing Rules

• MUST use httpx.AsyncClient for async tests
• MUST override dependencies for test isolation
• SHOULD use in-memory database for unit tests
• MUST test all response status codes

---

Background Tasks

from fastapi import BackgroundTasks

async def send_notification(email: str, message: str) -> None:
    # Async notification logic
    ...

@router.post("/users/")
async def create_user(
    user: UserCreate,
    background_tasks: BackgroundTasks,
) -> UserResponse:
    new_user = await create_user_in_db(user)
    background_tasks.add_task(send_notification, user.email, "Welcome!")
    return new_user

Background Task Rules

• SHOULD use for non-blocking operations (email, notifications)
• MUST NOT use for critical operations requiring confirmation
• For complex jobs, SHOULD use Celery or similar task queue

---

Middleware

from fastapi import FastAPI, Request
from starlette.middleware.base import BaseHTTPMiddleware
import time

class TimingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        start = time.perf_counter()
        response = await call_next(request)
        duration = time.perf_counter() - start
        response.headers["X-Process-Time"] = str(duration)
        return response

app.add_middleware(TimingMiddleware)

# CORS middleware
from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=settings.allowed_origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

Middleware Rules

• MUST add CORS middleware for browser clients
• SHOULD add request timing for observability
• Middleware MUST NOT block the event loop
• Order matters: add most critical middleware last

---

OpenAPI Documentation

from fastapi import FastAPI

app = FastAPI(
    title="My API",
    description="API for managing resources",
    version="1.0.0",
    terms_of_service="https://example.com/terms/",
    contact={
        "name": "API Support",
        "url": "https://example.com/support",
        "email": "[e-mail]",
    },
    license_info={
        "name": "MIT",
        "url": "https://opensource.org/licenses/MIT",
    },
)

# Route documentation
@router.get(
    "/{user_id}",
    summary="Get a user by ID",
    description="Retrieve detailed user information by their unique identifier.",
    response_description="The user object",
    responses={
        404: {"description": "User not found"},
        422: {"description": "Validation error"},
    },
)
async def get_user(user_id: int) -> UserResponse:
    """
    Get a user with all their details.

    - **user_id**: The unique identifier of the user
    """
    ...

Documentation Rules

• MUST provide summary and description for all routes
• SHOULD document all possible response codes
• MUST use docstrings for detailed parameter descriptions
• Schemas SHOULD have Field descriptions

---

Security Patterns

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

async def get_current_user(
    token: str = Depends(oauth2_scheme),
    db: AsyncSession = Depends(get_db),
) -> User:
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, settings.secret_key, algorithms=["HS256"])
        user_id: int = payload.get("sub")
        if user_id is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception

    user = await db.get(User, user_id)
    if user is None:
        raise credentials_exception
    return user

# Protected route
@router.get("/me")
async def read_users_me(current_user: User = Depends(get_current_user)) -> UserResponse:
    return current_user

Security Rules

• MUST use OAuth2 or API key authentication for protected routes
• MUST validate and decode tokens in dependencies
• MUST NOT store plaintext passwords
• SHOULD use HTTPS in production