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="user@example.com")
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"