fastapi

name	fastapi
description	FastAPI patterns for building high-performance Python APIs. Covers routing, dependency injection, Pydantic models, background tasks, WebSockets, testing, and production deployment.

FastAPI Skill

Modern FastAPI patterns for building high-performance Python APIs.

Quick Start

Installation

# pip
pip install fastapi uvicorn[standard]

# poetry
poetry add fastapi uvicorn[standard]

# uv
uv add fastapi uvicorn[standard]

Run Development Server

uvicorn app.main:app --reload --host 0.0.0.0 --port 8000

Project Structure

app/
├── __init__.py
├── main.py              # FastAPI app entry
├── config.py            # Settings/configuration
├── database.py          # DB connection
├── models/              # SQLModel/SQLAlchemy models
│   ├── __init__.py
│   └── task.py
├── schemas/             # Pydantic schemas
│   ├── __init__.py
│   └── task.py
├── routers/             # API routes
│   ├── __init__.py
│   └── tasks.py
├── services/            # Business logic
│   ├── __init__.py
│   └── task_service.py
├── dependencies/        # Shared dependencies
│   ├── __init__.py
│   └── auth.py
└── tests/
    └── test_tasks.py

Key Concepts

Concept	Guide
Routing	reference/routing.md
Dependencies	reference/dependencies.md
Pydantic Models	reference/pydantic.md
Background Tasks	reference/background-tasks.md
WebSockets	reference/websockets.md

Examples

Pattern	Guide
CRUD Operations	examples/crud.md
Authentication	examples/authentication.md
File Upload	examples/file-upload.md
Testing	examples/testing.md

Templates

Template	Purpose
templates/main.py	App entry point
templates/router.py	Router template
templates/config.py	Settings with Pydantic

Basic App

# app/main.py
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

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

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

@app.get("/health")
async def health():
    return {"status": "healthy"}

Routers

# app/routers/tasks.py
from fastapi import APIRouter, Depends, HTTPException, status
from sqlmodel import Session, select
from app.database import get_session
from app.models import Task
from app.schemas import TaskCreate, TaskRead, TaskUpdate
from app.dependencies.auth import get_current_user, User

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


@router.get("", response_model=list[TaskRead])
async def get_tasks(
    user: User = Depends(get_current_user),
    session: Session = Depends(get_session),
):
    statement = select(Task).where(Task.user_id == user.id)
    return session.exec(statement).all()


@router.post("", response_model=TaskRead, status_code=status.HTTP_201_CREATED)
async def create_task(
    task_data: TaskCreate,
    user: User = Depends(get_current_user),
    session: Session = Depends(get_session),
):
    task = Task(**task_data.model_dump(), user_id=user.id)
    session.add(task)
    session.commit()
    session.refresh(task)
    return task


@router.get("/{task_id}", response_model=TaskRead)
async def get_task(
    task_id: int,
    user: User = Depends(get_current_user),
    session: Session = Depends(get_session),
):
    task = session.get(Task, task_id)
    if not task or task.user_id != user.id:
        raise HTTPException(status_code=404, detail="Task not found")
    return task


@router.patch("/{task_id}", response_model=TaskRead)
async def update_task(
    task_id: int,
    task_data: TaskUpdate,
    user: User = Depends(get_current_user),
    session: Session = Depends(get_session),
):
    task = session.get(Task, task_id)
    if not task or task.user_id != user.id:
        raise HTTPException(status_code=404, detail="Task not found")

    for key, value in task_data.model_dump(exclude_unset=True).items():
        setattr(task, key, value)

    session.add(task)
    session.commit()
    session.refresh(task)
    return task


@router.delete("/{task_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_task(
    task_id: int,
    user: User = Depends(get_current_user),
    session: Session = Depends(get_session),
):
    task = session.get(Task, task_id)
    if not task or task.user_id != user.id:
        raise HTTPException(status_code=404, detail="Task not found")
    session.delete(task)
    session.commit()

Dependency Injection

# app/dependencies/auth.py
from fastapi import Depends, HTTPException, Header
from dataclasses import dataclass

@dataclass
class User:
    id: str
    email: str

async def get_current_user(
    authorization: str = Header(..., alias="Authorization")
) -> User:
    # Verify JWT token
    # ... verification logic ...
    return User(id="user_123", email="[e-mail]")


def require_role(role: str):
    async def checker(user: User = Depends(get_current_user)):
        if user.role != role:
            raise HTTPException(status_code=403, detail="Forbidden")
        return user
    return checker

Pydantic Schemas

# app/schemas/task.py
from pydantic import BaseModel, Field
from datetime import datetime
from typing import Optional


class TaskCreate(BaseModel):
    title: str = Field(..., min_length=1, max_length=200)
    description: Optional[str] = None


class TaskUpdate(BaseModel):
    title: Optional[str] = Field(None, min_length=1, max_length=200)
    description: Optional[str] = None
    completed: Optional[bool] = None


class TaskRead(BaseModel):
    id: int
    title: str
    description: Optional[str]
    completed: bool
    user_id: str
    created_at: datetime
    updated_at: datetime

    model_config = {"from_attributes": True}

Background Tasks

from fastapi import BackgroundTasks

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

@router.post("/notify")
async def notify(
    email: str,
    background_tasks: BackgroundTasks,
):
    background_tasks.add_task(send_email, email, "Hello!")
    return {"message": "Notification queued"}

Configuration

# app/config.py
from pydantic_settings import BaseSettings
from functools import lru_cache


class Settings(BaseSettings):
    database_url: str
    better_auth_url: str = "http://localhost:3000"
    debug: bool = False

    model_config = {"env_file": ".env"}


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

Error Handling

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


class AppException(Exception):
    def __init__(self, status_code: int, detail: str):
        self.status_code = status_code
        self.detail = detail


@app.exception_handler(AppException)
async def app_exception_handler(request: Request, exc: AppException):
    return JSONResponse(
        status_code=exc.status_code,
        content={"detail": exc.detail},
    )

Testing

# tests/test_tasks.py
import pytest
from fastapi.testclient import TestClient
from app.main import app

client = TestClient(app)


def test_health():
    response = client.get("/health")
    assert response.status_code == 200
    assert response.json() == {"status": "healthy"}


def test_create_task(auth_headers):
    response = client.post(
        "/api/tasks",
        json={"title": "Test task"},
        headers=auth_headers,
    )
    assert response.status_code == 201
    assert response.json()["title"] == "Test task"