Overview
Modern web APIs need to be fast, reliable, and easy to maintain. FastAPI (https://fastapi.tiangolo.com/) combines Python’s type hints with an async-first design to deliver high-performance, developer-friendly APIs. This guide covers three fundamental concepts you should master: asynchronous Python, API endpoint design, and FastAPI’s dependency injection system. Each section includes concise explanations, production-quality examples, and practical tips.
Async Python — what it is and why it matters
What async is
- Asyncio (the
async/awaitmodel) is Python’s approach for non-blocking concurrency. It’s ideal for I/O-bound workloads (HTTP calls, DB queries, file I/O) because it lets a single thread manage many concurrent tasks by yielding control on awaits. - Use async for handling many simultaneous requests without spawning many threads or processes; avoid blocking the event loop with CPU-heavy synchronous work.
Practical example — concurrent HTTP requests
The example below demonstrates an async endpoint that concurrently fetches multiple URLs using httpx.AsyncClient. It is runnable and shows how to structure async paths in FastAPI.
# content: examples/async_fetch.py
from typing import List
import httpx
from fastapi import FastAPI, HTTPException
app = FastAPI()
async def fetch(client: httpx.AsyncClient, url: str) -> dict:
resp = await client.get(url)
resp.raise_for_status()
return {"url": url, "status": resp.status_code}
@app.get("/fetch")
async def fetch_urls(urls: List[str] = []):
"""Fetch multiple URLs concurrently.
Provide `?urls=https://a.com&urls=https://b.com` or a single `urls` query param.
"""
if not urls:
raise HTTPException(status_code=400, detail="urls parameter is required")
# Limit concurrency to avoid overwhelming remote servers and connection pools
sem = asyncio.Semaphore(10)
async def bounded_fetch(u: str):
async with sem:
return await fetch(client, u)
async with httpx.AsyncClient(timeout=10.0, limits=httpx.Limits(max_connections=20)) as client:
tasks = [bounded_fetch(u) for u in urls]
results = await asyncio.gather(*tasks)
return {"results": results}
Notes:
- Use
asyncview functions for non-blocking I/O. - Prefer libraries with native async support (e.g.,
httpx, async DB drivers). - Avoid
time.sleep()and other blocking calls; useawait asyncio.sleep()instead.
Core terms & abbreviations
- ASGI — Asynchronous Server Gateway Interface: the async successor to WSGI used by frameworks like FastAPI; servers include Uvicorn and Hypercorn. (ASGI spec)
- Uvicorn — A fast ASGI server commonly used to run FastAPI applications. (https://www.uvicorn.org/)
- Pydantic — Data validation & settings management library used by FastAPI for request/response models. (https://pydantic-docs.helpmanual.io/)
- SQLModel — A library from the FastAPI author combining Pydantic & SQLAlchemy patterns for typing-friendly DB models. (https://sqlmodel.tiangolo.com/)
- DI — Dependency Injection: a pattern to provide components (DB, clients, auth) via
Dependsin FastAPI. - JWT — JSON Web Token, a compact, URL-safe token for stateless auth. (https://jwt.io/)
Quick reference: ASGI, Uvicorn, Pydantic, SQLModel, DI (Dependency Injection), JWT.
Handling blocking (CPU-bound) work
If you must run blocking code (e.g., CPU-heavy tasks), offload it to a thread/process pool with asyncio.get_event_loop().run_in_executor or use a background job queue (Celery, RQ, or Prefect).
import asyncio
from concurrent.futures import ThreadPoolExecutor
executor = ThreadPoolExecutor(max_workers=4)
async def blocking_work(x: int) -> int:
loop = asyncio.get_running_loop()
result = await loop.run_in_executor(executor, lambda: x ** 2)
return result
Async pitfalls & best practices
- Don’t call blocking I/O in async code (library calls using sync sockets,
requests,time.sleep). - Keep the event loop responsive: avoid long-running sync loops.
- Monitor event loop latency in production; alerts help catch blocking operations early.
API Endpoints — building robust, typed routes
What API endpoints are
- FastAPI maps Python function signatures and type hints to request parsing, input validation, and OpenAPI docs automatically.
- Use Pydantic models for request/response validation and
response_modelfor consistent output shapes.
Example — users API with Pydantic
This small example shows common patterns: path params, query params, request body validation, response models, and proper status codes.
# content: examples/users_api.py
from typing import Optional, List
from pydantic import BaseModel, EmailStr
from fastapi import FastAPI, HTTPException, status, Query, Path
class UserIn(BaseModel):
email: EmailStr
password: str
class UserOut(BaseModel):
id: int
email: EmailStr
app = FastAPI()
# In-memory store for example purposes
_users: List[UserOut] = []
@app.post("/users", response_model=UserOut, status_code=status.HTTP_201_CREATED)
async def create_user(payload: UserIn):
# NOTE: never store raw passwords in production; hash them with bcrypt/argon2
user = UserOut(id=len(_users) + 1, email=payload.email)
_users.append(user)
return user
@app.get("/users", response_model=List[UserOut])
async def list_users(limit: int = Query(10, ge=1, le=100), offset: int = 0):
return _users[offset : offset + limit]
@app.get("/users/{user_id}", response_model=UserOut)
async def get_user(user_id: int = Path(..., gt=0)):
try:
return _users[user_id - 1]
except IndexError:
raise HTTPException(status_code=404, detail="User not found")
### Async DB example (SQLModel + async session)
This example shows how to expose an async session as a dependency and query data using SQLModel (async SQLAlchemy engine).
```python
from typing import AsyncGenerator
from sqlmodel import SQLModel, Field, select
from sqlmodel.ext.asyncio.session import AsyncSession
from sqlalchemy.ext.asyncio import create_async_engine, AsyncEngine
DATABASE_URL = "sqlite+aiosqlite:///./test.db"
engine: AsyncEngine = create_async_engine(DATABASE_URL, echo=False, future=True)
class User(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
email: str
async def init_db():
async with engine.begin() as conn:
await conn.run_sync(SQLModel.metadata.create_all)
async def get_session() -> AsyncGenerator[AsyncSession, None]:
async with AsyncSession(engine) as session:
yield session
@app.on_event("startup")
async def on_startup():
await init_db()
@app.get("/db/users", response_model=List[User])
async def db_list_users(session: AsyncSession = Depends(get_session)):
result = await session.execute(select(User))
users = result.scalars().all()
return users
```
Notes:
- Use async DB drivers (e.g., `asyncpg` for Postgres) in production.
- Ensure pool sizes and connection limits are configured to match concurrency needs.
Notes & patterns:
- Always validate request bodies with Pydantic models (
UserIn). - Use
statusconstants andresponse_modelto control output & docs. - Document constraints using
Query,Path, andBodyparameters.
Best practices
- Validate input and sanitize outputs.
- Keep functions small and pure; delegate business logic to service layers.
- Use background tasks (
BackgroundTasks) for non-critical work like emails.
Dependency Injection (DI) — clean, testable components
What dependency injection is
- FastAPI’s DI uses
Dependsto declare dependencies in function signatures. Dependencies can be sync or async and can be used to provide services like DB sessions, authentication, configuration, or clients. - Dependencies improve testability (they can be overridden during tests) and separation of concerns.
Simple dependency example (auth)
# content: examples/deps_auth.py
from fastapi import FastAPI, Depends, HTTPException, status
from typing import Optional
app = FastAPI()
async def get_current_user(authorization: Optional[str] = None):
# In real code, read header via Header(...) or use OAuth2 schemes
if authorization != "Bearer secrettoken":
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")
return {"username": "alice"}
@app.get("/me")
async def me(user=Depends(get_current_user)):
return {"user": user}
### OAuth2 (password flow) + JWT example
This is a concise example showing `OAuth2PasswordBearer`, a token endpoint, and a `get_current_user` dependency that validates a JWT-like token.
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm from datetime import datetime, timedelta import jwt
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/token") SECRET_KEY = “change-this-secret”
def create_token(data: dict, expires_delta: timedelta | None = None) -> str: to_encode = data.copy() expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15)) to_encode.update({“exp”: expire}) return jwt.encode(to_encode, SECRET_KEY, algorithm=“HS256”)
@app.post(’/token’) async def token(form_data: OAuth2PasswordRequestForm = Depends()): # Validate user (demo only) if form_data.username != ‘alice’ or form_data.password != ‘wonderland’: raise HTTPException(status_code=400, detail=‘Incorrect username or password’) token = create_token({“sub”: form_data.username}) return {“access_token”: token, “token_type”: “bearer”}
async def get_current_user_token(token: str = Depends(oauth2_scheme)): try: payload = jwt.decode(token, SECRET_KEY, algorithms=[“HS256”]) return {“username”: payload.get(‘sub’)} except jwt.PyJWTError: raise HTTPException(status_code=401, detail=‘Invalid token’)
@app.get(’/me2’) async def me2(user=Depends(get_current_user_token)): return {“user”: user}
Notes: In production use well-audited libs (e.g., `python-jose`, `Authlib`) and secure secret management.
### Background tasks
For non-critical work (emails, notifications, light processing) use `BackgroundTasks` to avoid blocking the request/response cycle:
from fastapi import BackgroundTasks
def send_welcome_email(email: str) -> None: # pretend to send an email (sync or delegate to worker) print(f"sending email to {email}")
@app.post(’/register’) async def register(email: str, background_tasks: BackgroundTasks): background_tasks.add_task(send_welcome_email, email) return {“status”: “ok”}
### WebSockets
FastAPI supports WebSocket endpoints for bidirectional communication (e.g., live updates, chat).
from fastapi import WebSocket
@app.websocket(’/ws’) async def websocket_endpoint(ws: WebSocket): await ws.accept() try: while True: data = await ws.receive_text() await ws.send_text(f"echo: {data}") except Exception: await ws.close()
### Resource lifecycle: database session via yield (cleanup)
Use `contextlib.asynccontextmanager` or a `yield` dependency to create and clean up resources (database transactions, network clients).
content: examples/deps_db.py
from fastapi import FastAPI, Depends from typing import AsyncGenerator from contextlib import asynccontextmanager
app = FastAPI()
class DummyAsyncDB: async def connect(self): print(“connect”)
async def disconnect(self):
print("disconnect")
async def fetch_user(self, user_id: int):
return {"id": user_id, "email": "[email protected]"}
@asynccontextmanager async def get_db() -> AsyncGenerator[DummyAsyncDB, None]: db = DummyAsyncDB() await db.connect() try: yield db finally: await db.disconnect()
@app.get("/db/users/{user_id}") async def user_profile(user_id: int, db: DummyAsyncDB = Depends(get_db)): return await db.fetch_user(user_id)
Notes:
- `yield`-based deps ensure cleanup runs after the request completes, even on errors.
- Dependencies can be nested; FastAPI resolves and caches dependencies per-request by default.
### Testing & overrides
- Use `TestClient` and `app.dependency_overrides` to inject test doubles for dependencies (e.g., mocked DB), making tests deterministic and fast.
from fastapi.testclient import TestClient from examples.deps_db import app, get_db
client = TestClient(app)
async def fake_db(): class FakeDB: async def fetch_user(self, user_id): return {“id”: user_id, “email”: “[email protected]”} yield FakeDB()
app.dependency_overrides[get_db] = fake_db res = client.get(’/db/users/1’) assert res.json()[’email’] == ’[email protected]’
### DI pitfalls & best practices
- Avoid storing mutable global state in dependencies unless it's readonly or thread-safe.
- Be mindful of per-request caching: by default, a dependency used multiple times in a single request is executed once and its value reused.
- Use `Depends` for cross-cutting concerns (auth, db session, settings) and keep handlers focused on request semantics.
---
## Putting it all together — a small, async FastAPI app
This example ties async work, endpoints, and DI together with a small service that fetches external data and uses a dependency-provided HTTP client.
content: examples/app_together.py
from typing import List import httpx from fastapi import FastAPI, Depends from contextlib import asynccontextmanager
app = FastAPI()
@asynccontextmanager async def get_http_client(): async with httpx.AsyncClient(timeout=10.0) as client: yield client
@app.get(’/multi-fetch’) async def multi_fetch(urls: List[str], client: httpx.AsyncClient = Depends(get_http_client)): tasks = [client.get(u) for u in urls] responses = await httpx.gather(*tasks) return {“statuses”: [r.status_code for r in responses]}
Run with:
pip install fastapi[all] httpx uvicorn examples.app_together:app –reload
---
## Deployment & architecture (text graphs)
- Simple deployment:
client(browser or mobile) -> CDN -> load balancer -> FastAPI (uvicorn workers) -> database
- Serverless / Functions approach:
git -> CI -> deploy to platform (Vercel, AWS Lambda, Cloudflare Workers) -> managed DB / storage
Deployment notes:
- Use ASGI servers (uvicorn, hypercorn) for FastAPI.
- For high throughput, run multiple worker processes (e.g., Gunicorn with Uvicorn workers) and tune connection pools for DB.
- Use observability: structured logs, metrics, tracing; track latency and errors.
## Tips for running in production
- Example Gunicorn + Uvicorn workers command:
gunicorn -k uvicorn.workers.UvicornWorker -w 4 examples.app_together:app
- A common heuristic for worker count: `workers = 2 * cpu_count + 1` for CPU-bound apps; tune for I/O-bound async apps based on performance testing.
- Configure connection pools for DB clients to match concurrency and avoid exhausting DB connections.
---
## Pros / Cons & alternatives
- Pros:
- High performance for I/O-bound workloads (async-first).
- Automatic validation & OpenAPI docs from type hints.
- Easy testing and dependency isolation via `Depends`.
- Cons:
- Async programming has an initial learning curve.
- Careful with mixing blocking libraries; prefer async-native drivers.
- Alternatives: Flask (simple, sync-first), Django REST Framework (feature-rich, synchronous), FastAPI + SQLModel/Databases for type-safe DB access.
**Alternatives:** Flask (simple, sync-first), Django REST Framework (feature-rich, synchronous), FastAPI + SQLModel/Databases for type-safe DB access.
### Alternatives & when to choose them
- **Flask** — great for simple and small services or when an ecosystem of extensions is preferred; sync-first.
- **Django REST Framework** — full-featured, batteries-included for large projects needing auth, admin, and conventions.
- **Go (Gin/Fiber)** — consider for extremely high throughput and a compiled single-binary deployment model.
- **Serverless (Vercel, Cloudflare Workers)** — ideal for scaling short-lived functions with low operational overhead.
---
## Further resources
- FastAPI docs — <https://fastapi.tiangolo.com/>
- Pydantic — <https://pydantic-docs.helpmanual.io/>
- AsyncIO docs — <https://docs.python.org/3/library/asyncio.html>
- httpx — <https://www.python-httpx.org/>
- SQLModel (type-safe ORM for FastAPI) — <https://sqlmodel.tiangolo.com/>
- RealPython Asyncio Guide — <https://realpython.com/async-io-python/>
---
## Conclusion
FastAPI brings together modern Python language features (type hints and `async`/`await`) with pragmatic API tooling (Pydantic validation, automatic docs, and dependency injection). Mastering async patterns, designing typed endpoints, and using `Depends` for clean DI will let you build fast, maintainable, and testable APIs. If you'd like, I can scaffold a small starter repo (`starter/python-fastapi-basic`) with examples, test coverage, and a GitHub Actions workflow next.
## Resources
- [MDN Web Docs](https://developer.mozilla.org/)
- [Web.dev](https://web.dev/)
- [Can I Use](https://caniuse.com/)
Comments