|$ curl https://forge-ai.dev/api/markdown?path=docs/python/fastapi

$cat docs/python-fastapi.md

updated Today·20-26 min read·published

Python FastAPI

◆Python◆FastAPI◆Web◆Intermediate🎯Free Tools

Introduction

FastAPI is a modern async web framework built on Starlette and Pydantic. You get OpenAPI docs, dependency injection, and excellent editor support for free.

ℹ

info

Install: pip install "fastapi[standard]" — includes uvicorn.

Routes & Path Operations

routes.py

Python

1	from fastapi import FastAPI, HTTPException, status
2	from pydantic import BaseModel, Field
3
4	app = FastAPI(title="Forge API", version="1.0.0")
5
6	class ItemIn(BaseModel):
7	name: str = Field(min_length=1, max_length=100)
8	price: float = Field(gt=0)
9
10	class Item(ItemIn):
11	id: int
12
13	DB: dict[int, Item] = {}
14
15	@app.get("/health")
16	async def health() -> dict[str, str]:
17	return {"status": "ok"}
18
19	@app.post("/items", status_code=status.HTTP_201_CREATED)
20	async def create_item(body: ItemIn) -> Item:
21	item = Item(id=len(DB) + 1, **body.model_dump())
22	DB[item.id] = item
23	return item
24
25	@app.get("/items/{item_id}")
26	async def get_item(item_id: int) -> Item:
27	if item_id not in DB:
28	raise HTTPException(status_code=404, detail="not found")
29	return DB[item_id]

Depends

depends.py

Python

1	from fastapi import Depends, FastAPI, Header, HTTPException
2	from typing import Annotated
3
4	app = FastAPI()
5
6	async def get_token(authorization: Annotated[str \| None, Header()] = None) -> str:
7	if not authorization or not authorization.startswith("Bearer "):
8	raise HTTPException(status_code=401, detail="unauthorized")
9	return authorization.removeprefix("Bearer ")
10
11	Token = Annotated[str, Depends(get_token)]
12
13	@app.get("/me")
14	async def me(token: Token) -> dict[str, str]:
15	return {"token_prefix": token[:4]}

Async Endpoints

Use async def for I/O. CPU-bound or blocking sync libraries should run via asyncio.to_thread or a sync def endpoint (FastAPI runs it in a threadpool).

async_ep.py

Python

1	import asyncio
2	import httpx
3	from fastapi import FastAPI
4
5	app = FastAPI()
6
7	@app.get("/upstream")
8	async def upstream() -> dict:
9	async with httpx.AsyncClient(timeout=5.0) as client:
10	r = await client.get("https://httpbin.org/json")
11	r.raise_for_status()
12	return r.json()
13
14	@app.get("/cpu")
15	def cpu_bound() -> dict[str, int]:
16	# sync def -> threadpool
17	return {"n": sum(range(100_000))}

OpenAPI & Docs

Interactive docs at /docs (Swagger UI) and /redoc. Schema at /openapi.json.

openapi_meta.py

Python

1	from fastapi import FastAPI
2	from pydantic import BaseModel, Field
3
4	app = FastAPI(
5	title="Forge",
6	description="Demo API",
7	contact={"name": "ForgeLearn"},
8	)
9
10	class Err(BaseModel):
11	detail: str
12
13	@app.get(
14	"/items/{item_id}",
15	response_model=dict,
16	responses={404: {"model": Err}},
17	tags=["items"],
18	summary="Get item",
19	)
20	async def get_item(item_id: int = Field(gt=0)) -> dict:
21	return {"id": item_id}

Middleware

middleware.py

Python

1	import time
2	from fastapi import FastAPI, Request
3	from starlette.middleware.cors import CORSMiddleware
4
5	app = FastAPI()
6	app.add_middleware(
7	CORSMiddleware,
8	allow_origins=["https://forgelearn.dev"],
9	allow_methods=["*"],
10	allow_headers=["*"],
11	)
12
13	@app.middleware("http")
14	async def add_timing(request: Request, call_next):
15	t0 = time.perf_counter()
16	response = await call_next(request)
17	response.headers["X-Process-Time"] = f"{time.perf_counter() - t0:.4f}"
18	return response

Testing with httpx / TestClient

test_api.py

Python

1	from fastapi.testclient import TestClient
2	# from routes import app
3
4	# client = TestClient(app)
5	# r = client.post("/items", json={"name": "x", "price": 1.0})
6	# assert r.status_code == 201
7	# assert r.json()["name"] == "x"

Query, Path, Body, Form

params.py

Python

1	from fastapi import FastAPI, Query, Path, Body
2	from typing import Annotated
3	from pydantic import BaseModel, Field
4
5	app = FastAPI()
6
7	class Filters(BaseModel):
8	q: str \| None = None
9	limit: int = Field(10, ge=1, le=100)
10
11	@app.get("/search")
12	async def search(
13	q: Annotated[str \| None, Query(min_length=1)] = None,
14	limit: Annotated[int, Query(ge=1, le=100)] = 10,
15	) -> dict:
16	return {"q": q, "limit": limit}
17
18	@app.put("/items/{item_id}")
19	async def put_item(
20	item_id: Annotated[int, Path(gt=0)],
21	body: Annotated[dict, Body()],
22	) -> dict:
23	return {"id": item_id, **body}

response_model & Status

response_model.py

Python

1	from fastapi import FastAPI, Response, status
2	from pydantic import BaseModel
3
4	app = FastAPI()
5
6	class UserOut(BaseModel):
7	id: int
8	name: str
9
10	class UserIn(UserOut):
11	password: str
12
13	@app.post("/users", response_model=UserOut, status_code=status.HTTP_201_CREATED)
14	async def create(user: UserIn) -> UserIn:
15	# password stripped by response_model
16	return user

BackgroundTasks

bg.py

Python

1	from fastapi import BackgroundTasks, FastAPI
2
3	app = FastAPI()
4
5	def write_log(msg: str) -> None:
6	print("log", msg)
7
8	@app.post("/notify")
9	async def notify(background_tasks: BackgroundTasks) -> dict[str, str]:
10	background_tasks.add_task(write_log, "sent")
11	return {"status": "queued"}

APIRouter

router.py

Python

1	from fastapi import APIRouter, FastAPI
2
3	router = APIRouter(prefix="/v1", tags=["v1"])
4
5	@router.get("/ping")
6	async def ping() -> dict[str, str]:
7	return {"pong": "v1"}
8
9	app = FastAPI()
10	app.include_router(router)

Lifespan & Resources

lifespan.py

Python

1	from contextlib import asynccontextmanager
2	from fastapi import FastAPI
3	import httpx
4
5	@asynccontextmanager
6	async def lifespan(app: FastAPI):
7	app.state.client = httpx.AsyncClient(timeout=10.0)
8	yield
9	await app.state.client.aclose()
10
11	app = FastAPI(lifespan=lifespan)
12
13	@app.get("/ext")
14	async def ext() -> dict:
15	r = await app.state.client.get("https://httpbin.org/get")
16	return {"status": r.status_code}

Error Handling

errors.py

Python

1	from fastapi import FastAPI, Request
2	from fastapi.responses import JSONResponse
3
4	app = FastAPI()
5
6	class NotFoundError(Exception):
7	def __init__(self, name: str):
8	self.name = name
9
10	@app.exception_handler(NotFoundError)
11	async def not_found_handler(request: Request, exc: NotFoundError):
12	return JSONResponse(status_code=404, content={"detail": f"{exc.name} not found"})

Security Schemes

security.py

Python

1	from fastapi import Depends, FastAPI, HTTPException
2	from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
3	from typing import Annotated
4
5	app = FastAPI()
6	bearer = HTTPBearer()
7
8	async def current_user(
9	creds: Annotated[HTTPAuthorizationCredentials, Depends(bearer)],
10	) -> str:
11	if creds.credentials != "secret":
12	raise HTTPException(status_code=401, detail="invalid token")
13	return "ada"
14
15	@app.get("/secure")
16	async def secure(user: Annotated[str, Depends(current_user)]) -> dict[str, str]:
17	return {"user": user}

File Uploads

upload.py

Python

1	from fastapi import FastAPI, File, UploadFile
2	from pathlib import Path
3
4	app = FastAPI()
5
6	@app.post("/upload")
7	async def upload(file: UploadFile = File(...)) -> dict:
8	dest = Path("uploads") / (file.filename or "blob")
9	dest.parent.mkdir(exist_ok=True)
10	dest.write_bytes(await file.read())
11	return {"saved": str(dest), "content_type": file.content_type}

$Blueprint — Engineering Documentation·Section ID: PYTHON-FASTAPI·Revision: 1.0

Community

Get help on Slack, Discord or VIP

Stuck on a guide? Join the community and ask.

Join Slack Join Discord Kaundal VIP

X GitHub YouTube Instagram Facebook Discord Slack Telegram Buy me a coffee Ko-fi Sponsor