跪拜 Guibai
← Back to the summary

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


1. What is FastAPI

FastAPI is a modern Python web framework created by Sebastián Ramírez in 2018. It has three core characteristics:

  1. Fast: Based on ASGI async, performance approaches Node.js / Go
  2. Simple: Type-annotation driven, less code to write
  3. 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)

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:

  1. Repeated code in every route
  2. Complex session closing logic on exceptions
  3. Can't mock during testing

Using Depends:

  1. One dependency reused across all routes
  2. Automatic exception handling (finally)
  3. 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:

  1. Parse user_id from the URL
  2. Convert to int (returns 422 if conversion fails)
  3. 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"),
):
    ...

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:

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:

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


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 Answer

user_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 Answer

After 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 Answer

Matches 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:

Comments

Top 1 from juejin.cn, machine-translated. The original thread is authoritative.

逍康

You've got some real skills [grin]