jawwad-ali 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:
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
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:
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
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
# 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
# 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
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
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
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
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
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
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 patternsdatabase.md- SQLAlchemy and async database patternstesting.md- Testing strategies and patterns
See the examples/ directory for working code:
crud_router.py- Complete CRUD router exampleauth_dependencies.py- Authentication dependency examplesmiddleware_examples.py- Common middleware patterns