Summary
FastAPI is a modern, fast web framework for building APIs with Python 3.7+ based on standard Python type hints. Built on Starlette and Pydantic, it provides automatic API documentation, data validation, serialization, and native async support. Key features include high performance, easy testing, standards-based (OpenAPI, JSON Schema), and production-ready code generation.
1. FastAPI Basics
What is FastAPI?
- Modern, fast web framework for building APIs with Python 3.7+
- Based on standard Python type hints
- Built on Starlette (web) and Pydantic (data validation)
- Automatic API documentation (Swagger UI, ReDoc)
Installation
pip install fastapi
pip install "uvicorn[standard]" # ASGI server
Basic Application
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
async def root():
return {"message": "Hello World"}
# Run: uvicorn main:app --reload
2. Core Concepts
Path Parameters
@app.get("/items/{item_id}")
async def read_item(item_id: int):
return {"item_id": item_id}
# Path parameter with enum
from enum import Enum
class ModelName(str, Enum):
alexnet = "alexnet"
resnet = "resnet"
@app.get("/models/{model_name}")
async def get_model(model_name: ModelName):
return {"model_name": model_name}
Query Parameters
@app.get("/items/")
async def read_items(skip: int = 0, limit: int = 10):
return {"skip": skip, "limit": limit}
# Optional query parameters
from typing import Optional
@app.get("/items/{item_id}")
async def read_item(item_id: str, q: Optional[str] = None):
return {"item_id": item_id, "q": q}
Request Body (Pydantic Models)
from pydantic import BaseModel
class Item(BaseModel):
name: str
price: float
is_offer: Optional[bool] = None
@app.post("/items/")
async def create_item(item: Item):
return item
# Multiple body parameters
class User(BaseModel):
username: str
full_name: Optional[str] = None
@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item, user: User):
return {"item_id": item_id, "item": item, "user": user}
Response Model
class ItemOut(BaseModel):
name: str
price: float
@app.post("/items/", response_model=ItemOut)
async def create_item(item: Item):
return item # Extra fields will be filtered out
3. Request Validation
Field Validation
from pydantic import BaseModel, Field
class Item(BaseModel):
name: str = Field(..., min_length=1, max_length=50)
price: float = Field(..., gt=0, le=1000)
tax: Optional[float] = Field(None, ge=0)
# Path parameter validation
from fastapi import Path
@app.get("/items/{item_id}")
async def read_item(
item_id: int = Path(..., gt=0, le=1000)
):
return {"item_id": item_id}
# Query parameter validation
from fastapi import Query
@app.get("/items/")
async def read_items(
q: Optional[str] = Query(None, min_length=3, max_length=50)
):
return {"q": q}
Custom Validators
from pydantic import validator
class Item(BaseModel):
name: str
price: float
@validator('price')
def price_must_be_positive(cls, v):
if v <= 0:
raise ValueError('Price must be positive')
return v
4. Dependency Injection
Basic Dependencies
from fastapi import Depends
async def common_parameters(q: Optional[str] = None, skip: int = 0, limit: int = 100):
return {"q": q, "skip": skip, "limit": limit}
@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):
return commons
# Class-based dependencies
class CommonQueryParams:
def __init__(self, q: Optional[str] = None, skip: int = 0, limit: int = 100):
self.q = q
self.skip = skip
self.limit = limit
@app.get("/items/")
async def read_items(commons: CommonQueryParams = Depends()):
return commons
Sub-dependencies
def query_extractor(q: Optional[str] = None):
return q
def query_or_cookie_extractor(
q: str = Depends(query_extractor),
last_query: Optional[str] = Cookie(None)
):
if not q:
return last_query
return q
@app.get("/items/")
async def read_query(query_or_default: str = Depends(query_or_cookie_extractor)):
return {"q_or_cookie": query_or_default}
5. Authentication & Security
Basic HTTP Authentication
from fastapi import HTTPException, status
from fastapi.security import HTTPBasic, HTTPBasicCredentials
security = HTTPBasic()
def get_current_username(credentials: HTTPBasicCredentials = Depends(security)):
if credentials.username != "admin" or credentials.password != "secret":
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid credentials",
headers={"WWW-Authenticate": "Basic"},
)
return credentials.username
@app.get("/users/me")
async def read_current_user(username: str = Depends(get_current_username)):
return {"username": username}
JWT Token Authentication
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from datetime import datetime, timedelta
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
SECRET_KEY = "secret-key"
ALGORITHM = "HS256"
def create_access_token(data: dict):
to_encode = data.copy()
expire = datetime.utcnow() + timedelta(minutes=15)
to_encode.update({"exp": expire})
return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
async def get_current_user(token: str = Depends(oauth2_scheme)):
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
username: str = payload.get("sub")
if username is None:
raise HTTPException(status_code=401, detail="Invalid token")
except JWTError:
raise HTTPException(status_code=401, detail="Invalid token")
return username
@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
# Verify username/password
access_token = create_access_token(data={"sub": form_data.username})
return {"access_token": access_token, "token_type": "bearer"}
6. Database Integration
SQLAlchemy Setup
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
# Dependency
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
Database Models
from sqlalchemy import Column, Integer, String, Float
class ItemDB(Base):
__tablename__ = "items"
id = Column(Integer, primary_key=True, index=True)
name = Column(String, index=True)
price = Column(Float)
# CRUD operations
from sqlalchemy.orm import Session
@app.post("/items/")
async def create_item(item: Item, db: Session = Depends(get_db)):
db_item = ItemDB(**item.dict())
db.add(db_item)
db.commit()
db.refresh(db_item)
return db_item
@app.get("/items/{item_id}")
async def read_item(item_id: int, db: Session = Depends(get_db)):
return db.query(ItemDB).filter(ItemDB.id == item_id).first()
7. Advanced Features
Middleware
from fastapi.middleware.cors import CORSMiddleware
import time
# CORS middleware (production configuration)
app.add_middleware(
CORSMiddleware,
allow_origins=["https://yourdomain.com"], # Specific origins in production
allow_credentials=True,
allow_methods=["GET", "POST", "PUT", "DELETE"],
allow_headers=["Authorization", "Content-Type"],
)
# 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_notification(email: str, message=""):
with open("log.txt", mode="w") as email_file:
content = f"notification for {email}: {message}"
email_file.write(content)
@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
background_tasks.add_task(write_notification, email, message="some notification")
return {"message": "Notification sent"}
WebSockets
from fastapi import WebSocket
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
while True:
data = await websocket.receive_text()
await websocket.send_text(f"Message: {data}")
File Upload
from fastapi import File, UploadFile
@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile = File(...)):
contents = await file.read()
return {"filename": file.filename, "size": len(contents)}
# Multiple files
@app.post("/uploadfiles/")
async def create_upload_files(files: List[UploadFile] = File(...)):
return {"filenames": [file.filename for file in files]}
8. Error Handling
HTTP Exceptions
from fastapi import HTTPException
@app.get("/items/{item_id}")
async def read_item(item_id: str):
if item_id not in items:
raise HTTPException(status_code=404, detail="Item not found")
return {"item": items[item_id]}
# Custom exception handlers
from fastapi import Request
from fastapi.responses import JSONResponse
class UnicornException(Exception):
def __init__(self, name: str):
self.name = name
@app.exception_handler(UnicornException)
async def unicorn_exception_handler(request: Request, exc: UnicornException):
return JSONResponse(
status_code=418,
content={"message": f"Oops! {exc.name} did something."}
)
9. Testing
Basic Testing with pytest
from fastapi.testclient import TestClient
client = TestClient(app)
def test_read_main():
response = client.get("/")
assert response.status_code == 200
assert response.json() == {"message": "Hello World"}
def test_create_item():
response = client.post(
"/items/",
json={"name": "Test Item", "price": 10.5}
)
assert response.status_code == 200
assert response.json()["name"] == "Test Item"
# Testing with dependencies override
from fastapi import Depends
def override_dependency():
return {"q": "overridden"}
app.dependency_overrides[common_parameters] = override_dependency
10. Performance & Best Practices
Async vs Sync Guidelines
| Operation Type | Use | Example | Reason |
|---|---|---|---|
| Database queries | async def |
await db.query() |
I/O bound operation |
| HTTP API calls | async def |
await httpx.get() |
Network I/O |
| File operations | async def |
await aiofiles.open() |
Disk I/O |
| CPU-intensive tasks | def |
Mathematical calculations | CPU bound |
| Simple operations | def |
Basic data transformations | No I/O involved |
# ✅ Good: Async for I/O operations
@app.get("/users/{user_id}")
async def get_user(user_id: int, db: Session = Depends(get_db)):
user = await db.query(User).filter(User.id == user_id).first()
return user
# ✅ Good: Sync for CPU-bound operations
@app.post("/calculate")
def calculate_result(data: CalculationInput):
result = complex_mathematical_operation(data.values)
return {"result": result}
# ❌ Avoid: Mixing sync/async incorrectly
@app.get("/bad-example")
async def bad_example():
# Don't call sync operations in async functions without proper handling
result = blocking_operation() # This blocks the event loop
return result
Response Caching
from fastapi import Response
@app.get("/cached")
async def get_cached_data(response: Response):
response.headers["Cache-Control"] = "public, max-age=3600"
return {"data": "This will be cached"}
Pagination
from typing import List
@app.get("/items/", response_model=List[Item])
async def read_items(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
items = db.query(ItemDB).offset(skip).limit(limit).all()
return items
11. Key Concepts & Comparisons
FastAPI vs Other Frameworks
| Framework | Performance | Documentation | Validation | Async Support |
|---|---|---|---|---|
| FastAPI | Very High | Automatic | Built-in (Pydantic) | Native |
| Flask | Medium | Manual | Manual | Plugin-based |
| Django | Medium | Manual | Built-in | Limited |
| Django REST | Medium | Manual | Built-in | Limited |
Dependency Injection System
| Concept | Description | Example |
|---|---|---|
| Basic Dependency | Function injected into path operation | Depends(get_db) |
| Sub-dependencies | Dependencies can have their own dependencies | Nested Depends() |
| Caching | Dependencies cached per request | Automatic |
| Override | Can override dependencies for testing | app.dependency_overrides |
Pydantic Integration
| Feature | Purpose | Benefit |
|---|---|---|
| Type Validation | Automatic data validation | Runtime type checking |
| Serialization | Convert between Python objects and JSON | Automatic conversion |
| Documentation | Generate JSON Schema | API docs generation |
| IDE Support | Type hints for autocompletion | Better developer experience |
Database Connection Patterns
# Generator pattern (recommended)
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
# Usage in endpoint
@app.get("/items/")
async def read_items(db: Session = Depends(get_db)):
return db.query(Item).all()
# Benefits:
# - Automatic connection cleanup
# - Dependency injection
# - Easy testing with overrides
Security Best Practices
| Practice | Implementation | Purpose |
|---|---|---|
| OAuth2 + JWT | OAuth2PasswordBearer |
Stateless authentication |
| HTTPS Only | Production deployment | Encrypt data in transit |
| Input Validation | Pydantic models | Prevent injection attacks |
| CORS Configuration | CORSMiddleware |
Control cross-origin requests |
| Rate Limiting | Third-party middleware | Prevent abuse |
| Secret Management | Environment variables | Protect sensitive data |
12. Quick Reference & Best Practices
HTTP Status Codes
| Code | Constant | Use Case |
|---|---|---|
| 200 | HTTP_200_OK |
Successful GET, PUT |
| 201 | HTTP_201_CREATED |
Successful POST |
| 204 | HTTP_204_NO_CONTENT |
Successful DELETE |
| 400 | HTTP_400_BAD_REQUEST |
Invalid request data |
| 401 | HTTP_401_UNAUTHORIZED |
Authentication required |
| 403 | HTTP_403_FORBIDDEN |
Access denied |
| 404 | HTTP_404_NOT_FOUND |
Resource not found |
| 422 | HTTP_422_UNPROCESSABLE_ENTITY |
Validation error |
| 500 | HTTP_500_INTERNAL_SERVER_ERROR |
Server error |
Request Components Reference
from fastapi import Request, Header, Cookie, Form, File, UploadFile
@app.post("/complete")
async def complete_request(
request: Request, # Full request object
item: Item, # JSON body (Pydantic model)
user_agent: Optional[str] = Header(None), # HTTP headers
session_id: Optional[str] = Cookie(None), # Cookies
username: str = Form(...), # Form data
file: UploadFile = File(...) # File upload
):
return {
"method": request.method,
"url": str(request.url),
"item": item,
"user_agent": user_agent
}
Essential Decorators & Lifecycle
| Decorator | Purpose | Example |
|---|---|---|
@app.get() |
Handle GET requests | @app.get("/items/") |
@app.post() |
Handle POST requests | @app.post("/items/") |
@app.on_event("startup") |
Run on app startup | Database initialization |
@app.on_event("shutdown") |
Run on app shutdown | Cleanup resources |
@app.middleware("http") |
Custom HTTP middleware | Logging, authentication |
@app.exception_handler() |
Handle specific exceptions | Custom error responses |
Development Best Practices
✅ Type Hints: Use Python type hints for all function parameters and return values
✅ Pydantic Models: Define clear data models for request/response validation
✅ Dependency Injection: Use Depends() for database connections, authentication
✅ Async Operations: Use async/await for I/O operations (database, HTTP calls)
✅ Error Handling: Implement proper exception handling with meaningful error messages
✅ Documentation: Leverage automatic OpenAPI documentation generation
Performance Optimization Tips
- Use
async/awaitfor I/O-bound operations - Implement database connection pooling
- Use response caching with appropriate headers
- Implement pagination for large datasets
- Use background tasks for non-critical operations
- Monitor and profile API performance
Testing Strategy
from fastapi.testclient import TestClient
import pytest
# Basic test
def test_read_main():
with TestClient(app) as client:
response = client.get("/")
assert response.status_code == 200
# Test with dependency override
def test_with_db_override():
def override_get_db():
return test_db_session
app.dependency_overrides[get_db] = override_get_db
# Run tests
app.dependency_overrides.clear()