Request Handling
Query Parameters
Basic Query Parameters
@app.get("/items/")
async def read_items(skip: int = 0, limit: int = 10):
return {"skip": skip, "limit": limit}
# With type hints and defaults
@app.get("/items/")
async def read_items(
skip: int = 0,
limit: int = 10,
q: str = None,
active: bool = True
):
return {"skip": skip, "limit": limit, "q": q, "active": active}
Query Parameter Validation
from fastapi import Query
@app.get("/items/")
async def read_items(
q: str = Query(None, min_length=3, max_length=50, regex="^[a-zA-Z0-9_]+$"),
skip: int = Query(0, ge=0, le=100),
limit: int = Query(10, gt=0, le=100)
):
return {"q": q, "skip": skip, "limit": limit}
# Required query parameters
@app.get("/items/")
async def read_items(q: str = Query(..., min_length=3)):
return {"q": q}
# Multiple values for same parameter
@app.get("/items/")
async def read_items(q: List[str] = Query(None)):
return {"q": q}
Query Parameter Metadata
@app.get("/items/")
async def read_items(
q: str = Query(
None,
title="Query string",
description="Query string for the items to search",
min_length=3,
max_length=50,
regex="^[a-zA-Z0-9_]+$",
deprecated=True
)
):
return {"q": q}
Path Parameters
Basic Path Parameters
@app.get("/items/{item_id}")
async def read_item(item_id: int):
return {"item_id": item_id}
@app.get("/users/{user_id}/items/{item_id}")
async def read_user_item(user_id: int, item_id: str):
return {"user_id": user_id, "item_id": item_id}
Path Parameter Validation
from fastapi import Path
@app.get("/items/{item_id}")
async def read_item(
item_id: int = Path(..., title="The ID of the item", gt=0, le=1000)
):
return {"item_id": item_id}
# String path parameters with validation
@app.get("/items/{item_id}")
async def read_item(
item_id: str = Path(..., min_length=1, max_length=10, regex="^[a-zA-Z0-9]+$")
):
return {"item_id": item_id}
Path Parameter Ordering
# Order matters when mixing path and query parameters
@app.get("/items/{item_id}")
async def read_item(
item_id: int = Path(..., title="The ID of the item", gt=0),
q: str = None,
size: float = Query(..., gt=0)
):
return {"item_id": item_id, "q": q, "size": size}
Request Body
Basic Request Body
from pydantic import BaseModel
class Item(BaseModel):
name: str
description: str = None
price: float
tax: float = None
@app.post("/items/")
async def create_item(item: Item):
return item
Request Body with Path Parameters
@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item):
return {"item_id": item_id, **item.dict()}
Request Body with Query Parameters
@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item, q: str = None):
result = {"item_id": item_id, **item.dict()}
if q:
result.update({"q": q})
return result
Multiple Request Bodies
class User(BaseModel):
username: str
full_name: 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}
# Single value in body
from fastapi import Body
@app.put("/items/{item_id}")
async def update_item(
item_id: int,
item: Item,
user: User,
importance: int = Body(...)
):
return {"item_id": item_id, "item": item, "user": user, "importance": importance}
Request Body Validation
from pydantic import BaseModel, Field
from typing import Optional
class Item(BaseModel):
name: str = Field(..., min_length=1, max_length=100)
description: Optional[str] = Field(None, max_length=300)
price: float = Field(..., gt=0, description="The price must be greater than zero")
tax: Optional[float] = Field(None, ge=0)
@app.post("/items/")
async def create_item(item: Item):
return item
Headers
Reading Headers
from fastapi import Header
@app.get("/items/")
async def read_items(user_agent: str = Header(None)):
return {"User-Agent": user_agent}
# Convert hyphens to underscores
@app.get("/items/")
async def read_items(
user_agent: str = Header(None),
accept_language: str = Header(None)
):
return {"User-Agent": user_agent, "Accept-Language": accept_language}
# Disable automatic conversion
@app.get("/items/")
async def read_items(
strange_header: str = Header(None, convert_underscores=False)
):
return {"strange_header": strange_header}
Header Validation
@app.get("/items/")
async def read_items(
x_token: str = Header(..., min_length=10, max_length=50)
):
return {"X-Token": x_token}
# Multiple headers with same name
@app.get("/items/")
async def read_items(x_token: List[str] = Header(None)):
return {"X-Token values": x_token}
Cookies
Reading Cookies
from fastapi import Cookie
@app.get("/items/")
async def read_items(ads_id: str = Cookie(None)):
return {"ads_id": ads_id}
# Cookie validation
@app.get("/items/")
async def read_items(
session_id: str = Cookie(..., min_length=10, max_length=50)
):
return {"session_id": session_id}
Form Data
Basic Form Data
from fastapi import Form
@app.post("/login/")
async def login(username: str = Form(...), password: str = Form(...)):
return {"username": username}
# Form data with validation
@app.post("/login/")
async def login(
username: str = Form(..., min_length=3, max_length=20),
password: str = Form(..., min_length=8)
):
return {"username": username}
Form Data with Files
from fastapi import Form, File, UploadFile
@app.post("/files/")
async def create_file(
file: bytes = File(...),
token: str = Form(...)
):
return {"file_size": len(file), "token": token}
@app.post("/uploadfile/")
async def create_upload_file(
file: UploadFile = File(...),
token: str = Form(...)
):
return {"filename": file.filename, "token": token}
File Uploads
Single File Upload
from fastapi import File, UploadFile
@app.post("/files/")
async def create_file(file: bytes = File(...)):
return {"file_size": len(file)}
@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile = File(...)):
return {"filename": file.filename}
Multiple File Uploads
@app.post("/files/")
async def create_files(files: List[bytes] = File(...)):
return {"file_sizes": [len(file) for file in files]}
@app.post("/uploadfiles/")
async def create_upload_files(files: List[UploadFile] = File(...)):
return {"filenames": [file.filename for file in files]}
File Upload with Metadata
@app.post("/files/")
async def create_file(
file: UploadFile = File(...),
description: str = Form(...)
):
return {
"filename": file.filename,
"description": description,
"content_type": file.content_type
}
Processing Uploaded Files
@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile = File(...)):
contents = await file.read()
return {
"filename": file.filename,
"content_type": file.content_type,
"size": len(contents)
}
# Save uploaded file
import shutil
from pathlib import Path
@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile = File(...)):
with open(f"uploads/{file.filename}", "wb") as buffer:
shutil.copyfileobj(file.file, buffer)
return {"filename": file.filename}
Request Context
Accessing Request Object
from fastapi import Request
@app.get("/info")
async def get_info(request: Request):
return {
"client_host": request.client.host,
"method": request.method,
"url": str(request.url),
"headers": dict(request.headers),
"query_params": dict(request.query_params)
}
Request State
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
request.state.db = get_db()
response = await call_next(request)
return response
@app.get("/items/")
async def read_items(request: Request):
db = request.state.db
return {"items": db.get_items()}