FastAPI Under the Hood: ASGI, Dependency Injection, and the Lifespan Pattern
02 - FastAPI Framework Introduction and Principles
📦 GitHub Repository: https://github.com/2384830985/fastapi-production-demo
Article 2 in the series. This article explains what FastAPI is, why it's fast, how dependency injection works, how routes are matched, and how lifecycle management works.
What You'll Learn
- The essential difference between ASGI and WSGI
- The relationship between FastAPI, Starlette, and Pydantic
- The execution mechanism of dependency injection (
Depends) - Route matching and
APIRouterorganization lifespanlifecycle hooks (replacing the deprecatedon_event)- Why FastAPI is faster than Flask
1. What is FastAPI
FastAPI is a modern Python web framework created by Sebastián Ramírez in 2018. It has three core characteristics:
- Fast: Based on ASGI async, performance approaches Node.js / Go
- Simple: Type-annotation driven, less code to write
- Automatic Documentation: Swagger / ReDoc auto-generated
1.1 Three-Layer Dependency Relationship
FastAPI
│
├── Starlette (underlying ASGI framework, provides routing, middleware, HTTP)
│
└── Pydantic (data validation, turns type annotations into validation rules)
- Starlette: FastAPI's web kernel, a pure async ASGI framework
- Pydantic: Data validation library, FastAPI uses it to validate request bodies and serialize responses
- FastAPI: Adds a layer of syntactic sugar on top of the two, making development smoother
1.2 A Minimal FastAPI Application
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
def hello():
return {"msg": "hello"}
Startup:
uvicorn main:app --reload
main:app means the app variable in the main.py file.
2. ASGI vs WSGI: Why FastAPI is Fast
2.1 WSGI (Synchronous)
Older generation Python web frameworks (Flask, Django) use the WSGI protocol:
Request → WSGI Server (gunicorn) → Flask → Process → Response
Problem: One request occupies one worker. While waiting for the database, the worker is idle, wasting resources.
2.2 ASGI (Asynchronous)
Request → ASGI Server (uvicorn) → FastAPI (async) → Process (can yield CPU) → Response
Advantage: Asynchronous I/O, one worker can handle multiple requests. While waiting for the database, the CPU is yielded to other requests.
2.3 Synchronous vs Asynchronous Code Comparison
# Flask (synchronous)
@app.route("/users")
def get_users():
users = db.query_all() # Blocks waiting for DB, worker is idle
return jsonify(users)
# FastAPI (asynchronous)
@app.get("/users")
async def get_users():
users = await db.query_all() # Yields CPU while waiting for DB
return users
2.4 Why This Project Uses Synchronous Code
Look at the project code:
@router.get("")
def list_users(svc: UserService = Depends(get_user_service)): # def, not async def
return svc.list_users()
Using def instead of async def because SQLAlchemy 2.0's synchronous API is blocking. FastAPI automatically puts synchronous routes into a thread pool for execution, not blocking the event loop.
| Syntax | Suitable Scenario |
|---|---|
async def |
Async I/O libraries (asyncpg, httpx async, Redis async) |
def |
Synchronous libraries (SQLAlchemy sync, requests) |
This project uses SQLAlchemy's synchronous API, so routes use def.
3. Dependency Injection: The Principle of Depends
Dependency injection is FastAPI's most powerful feature. First, look at the usage:
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
@router.get("/users")
def list_users(db: Session = Depends(get_db)):
return db.query(User).all()
Depends(get_db) tells FastAPI: "Before calling list_users, first call get_db, and pass the result as the db parameter."
3.1 Execution Flow of yield Dependencies
def get_db():
db = SessionLocal() # 1. Create Session
try:
yield db # 2. Inject db into route
finally:
db.close() # 3. After route executes, close Session
Execution order:
Request comes in
↓
FastAPI calls get_db()
↓
get_db executes until yield, gives db to route
↓
Route executes (uses db to query database)
↓
Route returns
↓
FastAPI continues get_db's finally, closes db
↓
Returns response to client
Key Point: Whether the route succeeds or throws an exception, finally always executes, ensuring connection release.
3.2 Nested Dependencies
Dependencies can be nested:
def get_db():
db = SessionLocal()
yield db
def get_user_service(db: Session = Depends(get_db)):
return UserService(UserRepo(db))
@router.get("/users")
def list_users(svc: UserService = Depends(get_user_service)):
return svc.list_users()
FastAPI automatically resolves in dependency order:
get_db → gets db
↓
get_user_service(db) → gets svc
↓
list_users(svc)
3.3 Dependency Caching (Within the Same Request)
Within the same request, Depends(get_db) is only called once:
@router.get("/")
def index(
db1: Session = Depends(get_db),
db2: Session = Depends(get_db), # Same db instance
):
assert db1 is db2 # True
Benefit: One request shares one db session, avoiding duplicate creation.
3.4 Uses of Dependencies
| Scenario | What the Dependency Does |
|---|---|
| Database Session | Independent session per request |
| Current User | Parse current user from JWT |
| Permission Check | Check if authorized |
| Pagination Parameters | Parse skip/limit |
| Rate Limiting | Check call frequency |
3.5 Depends vs Plain Function Calls
Why not just call the function directly?
# ❌ Direct call
@router.get("/users")
def list_users():
db = SessionLocal() # Can't auto-close
try:
return db.query(User).all()
finally:
db.close()
Problems:
- Repeated code in every route
- Complex session closing logic on exceptions
- Can't mock during testing
Using Depends:
- One dependency reused across all routes
- Automatic exception handling (finally)
- Replace dependencies during testing:
app.dependency_overrides[get_db] = get_test_db
4. Route Matching and APIRouter
4.1 Route Decorators
@app.get("/") # GET request
@app.post("/users") # POST request
@app.put("/users/{id}") # PUT request
@app.delete("/users/{id}") # DELETE request
Each decorator corresponds to an HTTP method.
4.2 Path Parameters
@app.get("/users/{user_id}")
def get_user(user_id: int): # Type annotation auto-validates
return {"id": user_id}
user_id: int makes FastAPI automatically:
- Parse
user_idfrom the URL - Convert to int (returns 422 if conversion fails)
- Pass to the function
4.3 Query Parameters
@app.get("/users")
def list_users(skip: int = 0, limit: int = 20):
return {"skip": skip, "limit": limit}
Parameters not in the URL path (skip, limit) automatically become query parameters: /users?skip=0&limit=20.
4.4 Adding Constraints with Query
from fastapi import Query
@router.get("")
def list_users(
skip: int = Query(0, ge=0, description="Number of records to skip"),
limit: int = Query(20, ge=1, le=100, description="Number per page"),
):
...
ge=0: Must be ≥ 0le=100: Must be ≤ 100description: Displayed in Swagger documentation
4.5 APIRouter: Modular Routing
Large projects don't pile all routes into one file. Use APIRouter to split them:
# app/api/users.py
from fastapi import APIRouter
router = APIRouter(prefix="/users", tags=["User Management"])
@router.get("")
def list_users(): ...
@router.get("/{user_id}")
def get_user(user_id: int): ...
# main.py
from fastapi import FastAPI
from app.api import users_router
app = FastAPI()
app.include_router(users_router)
# All /users paths are handled by users_router
Benefits:
- Different resources split into files (users / orders / products)
prefixautomatically adds a prefix, avoiding repetitiontagsgroup displays in Swagger
4.6 Route Matching Order (Important Pitfall)
@router.get("/users/me") # ① Registered first
@router.get("/users/{user_id}") # ② Registered second
FastAPI matches in registration order. If reversed:
@router.get("/users/{user_id}") # Registered first
@router.get("/users/me") # Registered second
Accessing /users/me matches the first one, treating me as user_id, failing int conversion and returning 422.
Rule: Static paths first, dynamic paths second.
5. Lifecycle: lifespan vs on_event
5.1 Deprecated on_event
Old way:
@app.on_event("startup")
def on_startup():
Base.metadata.create_all(bind=engine)
@app.on_event("shutdown")
def on_shutdown():
engine.dispose()
Deprecated in FastAPI 0.93+, will be removed in the future.
5.2 New Way: lifespan
from contextlib import asynccontextmanager
@asynccontextmanager
async def lifespan(app: FastAPI):
# ── startup phase ──
Base.metadata.create_all(bind=engine)
yield # Suspend while application runs
# ── shutdown phase ──
engine.dispose()
app = FastAPI(lifespan=lifespan)
5.3 Principle of asynccontextmanager
asynccontextmanager turns a generator function into an async context manager.
@asynccontextmanager
async def lifespan(app):
print("startup") # Before yield = __aenter__
yield
print("shutdown") # After yield = __aexit__
Execution flow:
Application starts
↓
Call lifespan(app)
↓
Execute code before yield (startup)
↓
yield yields control, application starts receiving requests
↓
(Application runs, handles countless requests)
↓
Application receives shutdown signal (Ctrl+C / SIGTERM)
↓
Continue executing code after yield (shutdown)
↓
Application exits
5.4 Why Use lifespan Instead of on_event
| Dimension | on_event |
lifespan |
|---|---|---|
| Status | Deprecated | Recommended |
| Shared State | Troublesome (use global variables) | Simple (variables before yield can be passed) |
| Resource Management | Scattered (startup and shutdown are two functions) | Centralized (in one function) |
| Async Support | Average | Native |
5.5 This Project's lifespan
@asynccontextmanager
async def lifespan(app: FastAPI):
Base.metadata.create_all(bind=engine)
logger.info("Database tables ready")
yield
engine.dispose()
logger.info("Application closed, connection pool released")
app = FastAPI(lifespan=lifespan)
6. Request Processing Flow
Complete flow:
1. Client initiates HTTP request
↓
2. ASGI Server (uvicorn) receives
↓
3. Starlette middleware chain (CORS, authentication, etc.)
↓
4. FastAPI route matching
↓
5. Dependency injection resolution (Depends chain)
↓
6. Pydantic validates request (body / query / path parameters)
↓
7. Call route function
↓
8. Business logic execution (Service → Repository → DB)
↓
9. Pydantic serializes response (response_model)
↓
10. Middleware chain (response direction)
↓
11. ASGI Server returns response
6.1 Pydantic Validation Failure → 422
If the request body doesn't match the Schema, FastAPI automatically returns 422:
POST /users
{"username": "ab"} ← Length less than 3
Response 422:
{
"detail": [
{
"type": "string_too_short",
"loc": ["body", "username"],
"msg": "String should have at least 3 characters"
}
]
}
6.2 response_model Filters Fields
@router.post("", response_model=UserOut)
def create_user(payload: UserCreate, ...):
return svc.create_user(payload) # Returns UserOut
Even if the service returns UserInDB (containing hashed_password), FastAPI filters according to UserOut fields, double insurance against leaks.
7. Exception Handling
7.1 HTTPException
from fastapi import HTTPException, status
@router.get("/{user_id}")
def get_user(user_id: int):
user = db.get(user_id)
if user is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="User not found"
)
return user
FastAPI automatically converts HTTPException into the corresponding HTTP response.
7.2 Global Exception Handlers (Used in This Project)
@app.exception_handler(UserNotFoundError)
async def handle_user_not_found(_, exc):
return JSONResponse(status_code=404, content={"detail": str(exc)})
@app.exception_handler(Exception)
async def handle_unexpected_error(_, exc):
logger.exception(...)
return JSONResponse(status_code=500, content={"detail": "Internal server error"})
Benefit: Route functions don't need try/except, business exceptions are thrown directly.
7.3 status Constants
from fastapi import status
status.HTTP_200_OK # 200
status.HTTP_201_CREATED # 201
status.HTTP_204_NO_CONTENT # 204
status.HTTP_400_BAD_REQUEST # 400
status.HTTP_404_NOT_FOUND # 404
status.HTTP_409_CONFLICT # 409
status.HTTP_422_UNPROCESSABLE_ENTITY # 422
status.HTTP_500_INTERNAL_SERVER_ERROR # 500
Using constants is more readable than writing numbers directly.
8. FastAPI vs Flask Comprehensive Comparison
| Dimension | Flask | FastAPI |
|---|---|---|
| Protocol | WSGI (synchronous) | ASGI (asynchronous) |
| Performance | Slower | Fast (approaches Node.js) |
| Type Annotations | Optional | Core |
| Data Validation | Requires plugins | Built-in (Pydantic) |
| Auto Documentation | Requires flask-restx | Built-in Swagger + ReDoc |
| Async Support | Requires async-flask | Native |
| Learning Curve | Simple | Simple (type-annotation driven) |
| Ecosystem | Extremely rich | Rapidly growing |
8.1 Code Comparison for the Same Endpoint
Flask:
from flask import Flask, request, jsonify
app = Flask(__name__)
@app.route("/users", methods=["POST"])
def create_user():
data = request.get_json()
if not data.get("username") or len(data["username"]) < 3:
return jsonify({"error": "Username must be at least 3 characters"}), 400
if not data.get("password") or len(data["password"]) < 6:
return jsonify({"error": "Password must be at least 6 characters"}), 400
# ... manual validation
user = create_in_db(data)
return jsonify({"id": user.id, "username": user.username}), 201
FastAPI:
from fastapi import FastAPI
from pydantic import BaseModel, Field
app = FastAPI()
class UserCreate(BaseModel):
username: str = Field(..., min_length=3)
password: str = Field(..., min_length=6)
@app.post("/users", status_code=201)
def create_user(payload: UserCreate):
user = create_in_db(payload)
return {"id": user.id, "username": user.username}
FastAPI uses type annotations + Pydantic for automatic validation, cutting code volume by half.
9. Automatic Documentation
After starting the service, visit:
- http://127.0.0.1:8000/docs - Swagger UI (interactive testing)
- http://127.0.0.1:8000/redoc - ReDoc (read-only documentation)
9.1 Documentation Metadata
@router.post(
"",
response_model=UserOut,
status_code=status.HTTP_201_CREATED,
summary="Create user",
description="Create a new user. Username must be unique. Conflict returns 409.",
tags=["User Management"],
)
def create_user(...): ...
All this metadata is displayed in the Swagger documentation.
9.2 Why Use Documentation
- Frontend Integration: Give the frontend the Swagger URL directly, auto-generate API client
- Testing: Swagger UI can send requests directly
- Newcomer Onboarding: Reading docs is faster than reading code
10. Common Pitfalls
10.1 Synchronous Routes Blocking the Event Loop
# ❌ Synchronous blocking operation inside async def
@app.get("/slow")
async def slow():
time.sleep(5) # Blocks the entire event loop for 5 seconds
return {"ok": True}
Fix: Use def to let FastAPI put it in a thread pool, or use asyncio.sleep:
# ✅ Solution 1: Use def
@app.get("/slow")
def slow():
time.sleep(5)
return {"ok": True}
# ✅ Solution 2: Use async sleep
@app.get("/slow")
async def slow():
await asyncio.sleep(5)
return {"ok": True}
10.2 Forgetting await
# ❌ Forgot await
@app.get("/")
async def index():
result = some_async_func() # Returns coroutine, not executed
return result
# ✅ Correct
@app.get("/")
async def index():
result = await some_async_func()
return result
10.3 Route Order
# ❌ Dynamic route first
@router.get("/{user_id}")
@router.get("/me") # Will never be matched
# ✅ Static route first
@router.get("/me")
@router.get("/{user_id}")
10.4 Missing Fields in response_model
class UserOut(BaseModel):
id: int
username: str
@router.get("/{user_id}", response_model=UserOut)
def get_user(user_id: int):
return {"id": 1, "username": "alice", "password": "xxx"} # Extra field
# Actual response: {"id": 1, "username": "alice"} ← password filtered out
This is a good thing, but remember to define response_model properly.
11. Self-Test Questions
Q1: What's wrong with the following code?
@app.get("/users/{user_id}")
async def get_user(user_id):
return {"id": user_id}
View Answeruser_id has no type annotation, FastAPI doesn't know how to parse it. Should write user_id: int.
Q2: Depends(get_db) uses yield. When does finally execute?
View AnswerAfter the route function finishes executing (whether success or exception), FastAPI continues executing the code after yield, including the finally block.
Q3: Given the two routes below, which one matches when accessing /users/me?
@router.get("/users/{user_id}")
def get_user(user_id: int): ...
@router.get("/users/me")
def get_me(): ...
View AnswerMatches the first one /{user_id}, treats me as user_id, fails int conversion and returns 422. Should put /users/me first.
12. Summary
| Concept | Key Point |
|---|---|
| ASGI | Async protocol, yields CPU during I/O wait |
| FastAPI Architecture | Starlette (Web) + Pydantic (Validation) |
Depends |
Dependency injection, auto-call, auto-close, nestable |
APIRouter |
Modular routing, prefix adds prefix |
lifespan |
Replaces deprecated on_event, centralized lifecycle management |
response_model |
Auto serialization + field filtering |
| Global Exception Handling | Register with @app.exception_handler |
13. Next Article Preview
The next article covers Pydantic v2 Data Validation In-Depth: BaseModel principles, Field constraints, model_validate vs model_dump, ConfigDict, custom validators, generic models.
Further Reading:
Top 1 from juejin.cn, machine-translated. The original thread is authoritative.
You've got some real skills [grin]