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
Examples
Templates
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"
