FastAPI
FastAPI is a modern, fast web framework for building APIs with Python based on standard Python type hints.
Installation
Basic App
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
def read_root():
return {"message": "Hello World"}
@app.get("/items/{item_id}")
def read_item(item_id: int, q: str = None):
return {"item_id": item_id, "q": q}
# Run server
uvicorn main:app --reload
# Access at http://localhost:8000
# Auto docs at http://localhost:8000/docs
Path Parameters
@app.get("/users/{user_id}")
def get_user(user_id: int):
return {"user_id": user_id}
# Path with enum
from enum import Enum
class ModelName(str, Enum):
alexnet = "alexnet"
resnet = "resnet"
lenet = "lenet"
@app.get("/models/{model_name}")
def get_model(model_name: ModelName):
return {"model_name": model_name}
Query Parameters
# Optional parameters
@app.get("/items/")
def read_items(skip: int = 0, limit: int = 10):
return {"skip": skip, "limit": limit}
# Required parameters
@app.get("/search/")
def search(q: str):
return {"q": q}
# Multiple parameters
@app.get("/products/")
def get_products(
category: str = None,
min_price: float = 0,
max_price: float = 1000,
in_stock: bool = True
):
return {
"category": category,
"min_price": min_price,
"max_price": max_price,
"in_stock": in_stock
}
Request Body
from pydantic import BaseModel
class Item(BaseModel):
name: str
description: str = None
price: float
tax: float = None
@app.post("/items/")
def create_item(item: Item):
item_dict = item.dict()
if item.tax:
total = item.price + item.tax
item_dict.update({"total": total})
return item_dict
# Nested models
class User(BaseModel):
username: str
full_name: str = None
class ItemWithOwner(BaseModel):
name: str
price: float
owner: User
@app.post("/items/")
def create_item(item: ItemWithOwner):
return item
Response Model
from typing import List
class UserOut(BaseModel):
username: str
email: str
@app.get("/users/", response_model=List[UserOut])
def get_users():
users = [
{"username": "john", "email": "john@example.com", "password": "secret"},
{"username": "jane", "email": "jane@example.com", "password": "secret"},
]
return users # Password won't be included in response
Path Operations
@app.get("/items/")
def read_items():
return [{"name": "Item 1"}]
@app.post("/items/")
def create_item(item: Item):
return item
@app.put("/items/{item_id}")
def update_item(item_id: int, item: Item):
return {"item_id": item_id, **item.dict()}
@app.delete("/items/{item_id}")
def delete_item(item_id: int):
return {"message": "Item deleted"}
Status Codes
from fastapi import status
@app.post("/items/", status_code=status.HTTP_201_CREATED)
def create_item(item: Item):
return item
@app.delete("/items/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_item(item_id: int):
pass
HTTP Exceptions
from fastapi import HTTPException
@app.get("/items/{item_id}")
def read_item(item_id: int):
if item_id not in items:
raise HTTPException(
status_code=404,
detail="Item not found",
headers={"X-Error": "Custom header"}
)
return items[item_id]
Dependency Injection
from fastapi import Depends
# Simple dependency
def common_parameters(q: str = None, skip: int = 0, limit: int = 100):
return {"q": q, "skip": skip, "limit": limit}
@app.get("/items/")
def read_items(commons: dict = Depends(common_parameters)):
return commons
# Class dependency
class CommonQueryParams:
def __init__(self, q: str = None, skip: int = 0, limit: int = 100):
self.q = q
self.skip = skip
self.limit = limit
@app.get("/items/")
def read_items(commons: CommonQueryParams = Depends()):
return commons
Authentication
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
def get_current_user(token: str = Depends(oauth2_scheme)):
# Verify token
user = verify_token(token)
if not user:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid authentication credentials",
headers={"WWW-Authenticate": "Bearer"},
)
return user
@app.get("/users/me")
def read_users_me(current_user: User = Depends(get_current_user)):
return current_user
@app.post("/token")
def login(form_data: OAuth2PasswordRequestForm = Depends()):
user = authenticate_user(form_data.username, form_data.password)
if not user:
raise HTTPException(status_code=400, detail="Incorrect credentials")
access_token = create_access_token(data={"sub": user.username})
return {"access_token": access_token, "token_type": "bearer"}
CORS
from fastapi.middleware.cors import CORSMiddleware
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # Allows all origins
allow_credentials=True,
allow_methods=["*"], # Allows all methods
allow_headers=["*"], # Allows all headers
)
Background Tasks
from fastapi import BackgroundTasks
def write_log(message: str):
with open("log.txt", "a") as log:
log.write(message)
@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}")
return {"message": "Notification sent"}
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]
}
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}")
Database Integration (SQLAlchemy)
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(bind=engine)
Base = declarative_base()
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
email = Column(String, unique=True, index=True)
name = Column(String)
Base.metadata.create_all(bind=engine)
# Dependency
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
# Pydantic models
class UserCreate(BaseModel):
email: str
name: str
class UserResponse(BaseModel):
id: int
email: str
name: str
class Config:
orm_mode = True
# Endpoints
@app.post("/users/", response_model=UserResponse)
def create_user(user: UserCreate, db: Session = Depends(get_db)):
db_user = User(**user.dict())
db.add(db_user)
db.commit()
db.refresh(db_user)
return db_user
@app.get("/users/", response_model=List[UserResponse])
def get_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
users = db.query(User).offset(skip).limit(limit).all()
return users
APIRouter
# routers/users.py
from fastapi import APIRouter
router = APIRouter(
prefix="/users",
tags=["users"]
)
@router.get("/")
def get_users():
return [{"username": "john"}]
@router.get("/{user_id}")
def get_user(user_id: int):
return {"user_id": user_id}
# main.py
from routers import users
app.include_router(users.router)
Configuration
from pydantic import BaseSettings
class Settings(BaseSettings):
app_name: str = "My API"
debug: bool = False
database_url: str
class Config:
env_file = ".env"
settings = Settings()
@app.get("/info")
def info():
return {
"app_name": settings.app_name,
"debug": settings.debug
}
Testing
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", "price": 9.99}
)
assert response.status_code == 201
assert response.json()["name"] == "Test"
Deployment
# Install production server
pip install gunicorn
# Run with gunicorn
gunicorn main:app --workers 4 --worker-class uvicorn.workers.UvicornWorker --bind 0.0.0.0:8000