Request Body and Data Validation
Using Pydantic models to define request body.
from typing import Optional
from fastapi import FastAPI
from pydantic import BaseModel
class Item(BaseModel):
name: str
description: Optional[str] = None
price: float
tax: Optional[float] = None
app = FastAPI()
@app.post("/items/")
async def create_item(item: Item):
item_dict = item.dict()
if item.tax:
price_with_tax = item.price + item.tax
item_dict.update({"price_with_tax": price_with_tax})
return item_dict
|
FastAPI uses Pydantic for data validation. Type hints are automatically validated and serialized.
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, validator
class User(BaseModel):
id: int
name: str
signup_ts: Optional[datetime] = None
friends: List[int] = []
@validator('id')
def id_must_be_positive(cls, value):
if value <= 0:
raise ValueError('id must be positive')
return value
app = FastAPI()
@app.post("/users/")
async def create_user(user: User):
return user
|
|
from fastapi import FastAPI, HTTPException
app = FastAPI()
items = {"foo": "The Foo Wrestlers"}
@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 FastAPI, Request
from fastapi.responses import JSONResponse
class UnicornException(Exception):
def __init__(self, name: str):
self.name = name
app = FastAPI()
@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. There goes a unicorn :("},
)
@app.get("/unicorns/{unicorn_name}")
async def read_unicorn(unicorn_name: str):
if unicorn_name == "yolo":
raise UnicornException(name="Yolo")
return {"unicorn_name": unicorn_name}
|
Dependencies and Security
Using dependencies for reusable logic.
from typing import Optional
from fastapi import FastAPI, Depends
app = FastAPI()
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
@app.get("/users/")
async def read_users(commons: dict = Depends(common_parameters)):
return commons
|
OAuth2 with Password (and hashing)
|
from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext
# Example (simplified)
app = FastAPI()
@app.post("/token", response_model=Token)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
user = 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_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
access_token = create_access_token(
data={"sub": user.username},
expires_delta=access_token_expires
)
return {"access_token": access_token, "token_type": "bearer"}
|
|
from fastapi import Depends, FastAPI, HTTPException
from fastapi.security import APIKeyHeader
X_API_KEY = APIKeyHeader(name="X-API-Key")
async def get_api_key(api_key_header: str = Depends(X_API_KEY)):
if api_key_header == "YOUR_API_KEY":
return api_key_header
else:
raise HTTPException(status_code=400, detail="Invalid API Key")
app = FastAPI()
@app.get("/items/", dependencies=[Depends(get_api_key)])
async def read_items():
return [{"item": "Foo"}, {"item": "Bar"}]
|