FastAPI Development Guide

This skill provides comprehensive guidance for building high-performance Python APIs with FastAPI.

Core Concepts

Application Structure

FastAPI applications follow this recommended structure:

code

project/
├── app/
│   ├── __init__.py
│   ├── main.py              # Application entry point
│   ├── config.py            # Settings and configuration
│   ├── dependencies.py      # Shared dependencies
│   ├── models/              # Pydantic models
│   │   ├── __init__.py
│   │   └── user.py
│   ├── routers/             # API route handlers
│   │   ├── __init__.py
│   │   └── users.py
│   ├── services/            # Business logic
│   │   ├── __init__.py
│   │   └── user_service.py
│   └── db/                  # Database related
│       ├── __init__.py
│       └── database.py
├── tests/
├── requirements.txt
└── pyproject.toml

Basic Application Setup

python

from fastapi import FastAPI
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup: Initialize resources
    print("Starting up...")
    yield
    # Shutdown: Clean up resources
    print("Shutting down...")

app = FastAPI(
    title="My API",
    description="API description",
    version="1.0.0",
    lifespan=lifespan,
)

@app.get("/")
async def root():
    return {"message": "Hello World"}

Pydantic Models

Use Pydantic for request/response validation:

python

from pydantic import BaseModel, Field, EmailStr
from typing import Optional
from datetime import datetime

class UserBase(BaseModel):
    email: EmailStr
    username: str = Field(..., min_length=3, max_length=50)

class UserCreate(UserBase):
    password: str = Field(..., min_length=8)

class UserResponse(UserBase):
    id: int
    created_at: datetime
    is_active: bool = True

    model_config = {"from_attributes": True}

class UserUpdate(BaseModel):
    email: Optional[EmailStr] = None
    username: Optional[str] = Field(None, min_length=3, max_length=50)

Path Operations

python

from fastapi import FastAPI, Path, Query, Body, HTTPException, status

app = FastAPI()

@app.get("/users/{user_id}")
async def get_user(
    user_id: int = Path(..., gt=0, description="User ID"),
    include_posts: bool = Query(False, description="Include user posts"),
):
    user = await get_user_by_id(user_id)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="User not found"
        )
    return user

@app.post("/users/", status_code=status.HTTP_201_CREATED)
async def create_user(user: UserCreate):
    return await create_new_user(user)

@app.put("/users/{user_id}")
async def update_user(
    user_id: int,
    user: UserUpdate = Body(...),
):
    return await update_user_by_id(user_id, user)

@app.delete("/users/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_user(user_id: int):
    await delete_user_by_id(user_id)
    return None

APIRouter for Organization

python

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

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

@router.get("/", response_model=List[UserResponse])
async def list_users(skip: int = 0, limit: int = 100):
    return await get_users(skip=skip, limit=limit)

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

python

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

app = FastAPI()
app.include_router(users.router)
app.include_router(items.router, prefix="/api/v1")

Dependency Injection

python

from fastapi import Depends, HTTPException, status
from typing import Annotated

# Simple dependency
async def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# Dependency with parameters
def get_query_params(
    skip: int = 0,
    limit: int = Query(default=100, le=100),
):
    return {"skip": skip, "limit": limit}

# Using dependencies
@app.get("/items/")
async def get_items(
    db: Annotated[Session, Depends(get_db)],
    params: Annotated[dict, Depends(get_query_params)],
):
    return db.query(Item).offset(params["skip"]).limit(params["limit"]).all()

# Class-based dependency
class CommonParams:
    def __init__(self, skip: int = 0, limit: int = 100):
        self.skip = skip
        self.limit = limit

@app.get("/resources/")
async def get_resources(commons: Annotated[CommonParams, Depends()]):
    return {"skip": commons.skip, "limit": commons.limit}

Authentication with OAuth2

python

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from typing import Annotated
from jose import JWTError, jwt
from passlib.context import CryptContext

# Password hashing
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

# OAuth2 scheme
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

def verify_password(plain_password: str, hashed_password: str) -> bool:
    return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password: str) -> str:
    return pwd_context.hash(password)

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

    user = await get_user_by_username(username)
    if user is None:
        raise credentials_exception
    return user

async def get_current_active_user(
    current_user: Annotated[User, Depends(get_current_user)]
) -> User:
    if not current_user.is_active:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user

@app.post("/token")
async def login(form_data: Annotated[OAuth2PasswordRequestForm, Depends()]):
    user = await authenticate_user(form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token = create_access_token(data={"sub": user.username})
    return {"access_token": access_token, "token_type": "bearer"}

@app.get("/users/me")
async def read_users_me(
    current_user: Annotated[User, Depends(get_current_active_user)]
):
    return current_user

Error Handling

python

from fastapi import FastAPI, HTTPException, Request, status
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError

app = FastAPI()

# Custom exception
class ItemNotFoundError(Exception):
    def __init__(self, item_id: int):
        self.item_id = item_id

# Exception handler
@app.exception_handler(ItemNotFoundError)
async def item_not_found_handler(request: Request, exc: ItemNotFoundError):
    return JSONResponse(
        status_code=status.HTTP_404_NOT_FOUND,
        content={"detail": f"Item {exc.item_id} not found"},
    )

# Override validation error handler
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    return JSONResponse(
        status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
        content={
            "detail": exc.errors(),
            "body": exc.body,
        },
    )

# Usage
@app.get("/items/{item_id}")
async def get_item(item_id: int):
    item = await find_item(item_id)
    if not item:
        raise ItemNotFoundError(item_id)
    return item

Middleware

python

from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
import time

app = FastAPI()

# CORS middleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Custom middleware
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response

Background Tasks

python

from fastapi import BackgroundTasks

def write_log(message: str):
    with open("log.txt", "a") as f:
        f.write(f"{message}\n")

async def send_email(email: str, message: str):
    # Send email logic
    pass

@app.post("/send-notification/{email}")
async def send_notification(
    email: str,
    background_tasks: BackgroundTasks,
):
    background_tasks.add_task(write_log, f"Notification sent to {email}")
    background_tasks.add_task(send_email, email, "Welcome!")
    return {"message": "Notification sent in the background"}

Response Models and Status Codes

python

from fastapi import FastAPI, status
from fastapi.responses import JSONResponse, RedirectResponse

@app.post(
    "/items/",
    response_model=ItemResponse,
    status_code=status.HTTP_201_CREATED,
    responses={
        201: {"description": "Item created successfully"},
        400: {"description": "Invalid input"},
        409: {"description": "Item already exists"},
    },
)
async def create_item(item: ItemCreate):
    return await create_new_item(item)

@app.get("/redirect")
async def redirect():
    return RedirectResponse(url="/items/")

@app.get("/custom-response")
async def custom_response():
    return JSONResponse(
        content={"message": "Custom response"},
        headers={"X-Custom-Header": "value"},
    )

Best Practices

•Use Pydantic models for all request/response validation
•Organize with routers - Group related endpoints in separate router files
•Use dependency injection for shared logic (database, auth, etc.)
•Handle errors properly with appropriate status codes and messages
•Use async/await for I/O operations (database, HTTP calls)
•Add OpenAPI documentation via docstrings and response models
•Use type hints everywhere for better IDE support and validation
•Separate concerns - Keep business logic in services, not route handlers
•Use environment variables for configuration via pydantic-settings
•Write tests using TestClient from fastapi.testclient

Common Patterns

See the references/ directory for detailed patterns:

•authentication.md - OAuth2 and JWT authentication patterns
•database.md - SQLAlchemy and async database patterns
•testing.md - Testing strategies and patterns

See the examples/ directory for working code:

•crud_router.py - Complete CRUD router example
•auth_dependencies.py - Authentication dependency examples
•middleware_examples.py - Common middleware patterns