Dependency Injection System — The Core Design of FastAPI
Dependency injection is like the interfaces of LEGO bricks—each brick is concerned only with what it should provide and doesn't need to know who is using it. Database connections, the current user, and permission checks are automatically resolved through nested layers.
1. What You'll Learn
Depends()Basics: Function Dependencies, Class Dependencies, Nested Subdependencies- Dependency Lifecycle: Request-Level vs. Application-Level (
yieldDependencies and Resource Cleanup) - Global Dependencies and Routing Group Dependencies:
dependencies=[Depends(...)] - Dependency Coverage and Testing:
app.dependency_overridesPractical Tips - Alice Scenario: PriceTracker's DB session dependency, current user dependency, and permission check dependency—a three-layer nested DI chain
2. Alice's True Story
(1) Pain Point: Each endpoint repeatedly retrieves data from the database and user information
Alice's PriceTracker has 20 endpoints, each of which requires opening a database connection, verifying a JWT token, retrieving the current user, and checking permissions. She copies and pastes the same 15 lines of initialization code into each endpoint function, so if the database connection method changes, she has to make 20 separate modifications.
(2) Solutions for Dependency Injection
FastAPI's dependency injection extracts repetitive logic into reusable dependency functions. Endpoints only need to declare db = Depends(get_db), and FastAPI automatically resolves and injects them, including recursive resolution of sub-dependencies.
PYTHON
from fastapi import Depends
async def get_db():
db = Database()
yield db
db.close()
async def get_current_user(token: str, db=Depends(get_db)):
return verify_token(token, db)
@app.get("/products")
async def list_products(user=Depends(get_current_user), db=Depends(get_db)):
# user and db are auto-injected
...
(3) Revenue
Database connections, user authentication, and permission checks—which used to require 15 lines of code for each of the 20 endpoints—have been consolidated into three dependency functions, so a change made in one place takes effect globally. During testing, simply use app.dependency_overrides[get_db] = lambda: mock_db to replace the database information for all endpoints with a single line of code.
3. Depends() Basics
(1) Functional Dependencies
(1) ▶ Example: Simple Function Dependencies
PYTHON
from fastapi import FastAPI, Depends
app = FastAPI()
def common_parameters(
q: str | None = None,
skip: int = 0,
limit: int = 100,
):
return {"q": q, "skip": skip, "limit": limit}
@app.get("/products")
async def list_products(commons: dict = Depends(common_parameters)):
return commons
@app.get("/prices")
async def list_prices(commons: dict = Depends(common_parameters)):
return commons
Output:
TEXT
# Function defined successfully
(2) ▶ Example: Class Dependencies
PYTHON
from fastapi import FastAPI, Depends, Query
class CommonQueryParams:
def __init__(
self,
q: str | None = Query(None),
skip: int = Query(0, ge=0),
limit: int = Query(100, ge=1, le=200),
):
self.q = q
self.skip = skip
self.limit = limit
app = FastAPI()
@app.get("/products")
async def list_products(commons: CommonQueryParams = Depends(CommonQueryParams)):
return {"q": commons.q, "skip": commons.skip, "limit": commons.limit}
Output:
TEXT
# Function defined successfully
(2) Function Dependencies vs. Class Dependencies
| Dimension | Function Dependency | Class Dependency |
|---|---|---|
| Definition Method | def get_xxx() |
class Xxx: + __init__ |
| Use Cases | Simple logic, single call | Requires state, multiple methods |
| Parameter Analysis | Automatic Function Parameter Analysis | __init__ Automatic Parameter Analysis |
| Reusability | High | Medium |
| Recommendation Level | Top Choice | Use in Complex Scenarios |
4. Nested Subdependencies
(1) Dependency Resolution Tree
graph TD
A[get_current_user] --> B[get_db]
A --> C[get_token_from_header]
C --> D[OAuth2PasswordBearer]
E[require_admin] --> A
E --> F[check_subscription]
style A fill:#e1f5fe
style E fill:#fff3e0
FastAPI recursively parses the dependency tree: require_admin → get_current_user → get_db, with each dependency instantiated only once (cached within the same request).
(1) ▶ Example: Nested Subdependencies—DB → User
PYTHON
from fastapi import FastAPI, Depends, HTTPException, Header
app = FastAPI()
# Level 1: Database session
async def get_db():
db = {"connection": "active"}
try:
yield db
finally:
db["connection"] = "closed"
# Level 2: Get current user (depends on DB)
async def get_current_user(
authorization: str = Header(...),
db: dict = Depends(get_db),
):
if not authorization.startswith("Bearer "):
raise HTTPException(status_code=401, detail="Invalid token")
token = authorization.replace("Bearer ", "")
# Simplified: in production, verify JWT
user = {"id": 1, "username": "alice", "role": "admin"}
return user
# Level 3: Require admin (depends on current user)
async def require_admin(user: dict = Depends(get_current_user)):
if user["role"] != "admin":
raise HTTPException(status_code=403, detail="Admin required")
return user
@app.get("/admin/stats")
async def admin_stats(admin: dict = Depends(require_admin)):
return {"admin": admin["username"], "total_products": 1000000}
Output:
TEXT
# Function defined successfully
5. Dependency Lifecycle and yield
(1) Request-Level vs. Application-Level
| Lifecycle | Declaration Method | Scope | Typical Uses |
|---|---|---|---|
| Request Level | def dep(): yield x; cleanup |
Created and destroyed on every request | DB Session, Redis connection |
| Application Level | @app.on_event("startup") |
Created when the application starts | Engine initialization, connection pool |
(1) ▶ Example: yield Dependencies and Resource Cleanup
PYTHON
from fastapi import FastAPI, Depends
app = FastAPI()
# Request-scoped dependency with cleanup
async def get_db_session():
# Setup: create session
session = {"id": "session-123", "active": True}
print(f"DB session opened: {session['id']}")
try:
yield session # This is injected into the endpoint
finally:
# Cleanup: close session (runs after response)
session["active"] = False
print(f"DB session closed: {session['id']}")
@app.get("/products")
async def list_products(db: dict = Depends(get_db_session)):
print(f"Using DB session: {db['id']}")
return {"session_id": db["id"]}
Output (server logs):
TEXT
DB session opened: session-123
Using DB session: session-123
DB session closed: session-123
6. Global Dependencies and Route Group Dependencies
(1) Dependency Declarations with Different Scopes
(1) ▶ Example: Route Group Dependencies
PYTHON
from fastapi import FastAPI, Depends, APIRouter, Header, HTTPException
async def verify_api_key(x_api_key: str = Header(...)):
if x_api_key != "secret-key-123":
raise HTTPException(status_code=401, detail="Invalid API key")
return x_api_key
app = FastAPI()
# Public routes - no auth required
public_router = APIRouter()
@public_router.get("/health")
async def health():
return {"status": "healthy"}
# Protected routes - API key required for all routes in this group
protected_router = APIRouter(dependencies=[Depends(verify_api_key)])
@protected_router.get("/products")
async def list_products():
return [{"id": 1, "name": "Widget"}]
@protected_router.post("/products")
async def create_product():
return {"id": 2, "name": "New Product"}
# Register routers
app.include_router(public_router)
app.include_router(protected_router, prefix="/api/v1")
Output:
TEXT
# Function defined successfully
| Dependency Scope | Declaration Location | Scope of Effect |
|---|---|---|
| Endpoint Level | @app.get("/", dependencies=[...]) |
Single Endpoint |
| Route Group Level | APIRouter(dependencies=[...]) |
All endpoints in the route group |
| Global | FastAPI(dependencies=[...]) |
All application endpoints |
7. Dependency Coverage and Testing
(1) dependency_overrides
During testing, it is necessary to replace production dependencies (such as database connections and external APIs); app.dependency_overrides allows you to replace dependency implementations without modifying the source code.
(1) ▶ Example: Replacing Database Dependencies in Tests
PYTHON
from fastapi.testclient import TestClient
from fastapi import FastAPI, Depends
app = FastAPI()
# Real dependency
async def get_db():
return {"type": "postgresql", "host": "prod-db"}
# Mock dependency for testing
def get_mock_db():
return {"type": "sqlite", "host": "memory"}
@app.get("/db-info")
async def db_info(db: dict = Depends(get_db)):
return db
# In tests:
def test_db_info():
app.dependency_overrides[get_db] = get_mock_db
client = TestClient(app)
response = client.get("/db-info")
assert response.json() == {"type": "sqlite", "host": "memory"}
# Clean up
app.dependency_overrides.clear()
Output:
TEXT
# Function defined successfully
❓ FAQ
Q Are Depends dependencies called multiple times?
A Within the same request, each dependency is executed only once, and the result is cached for reuse. For different requests, the dependency is re-executed.
Q When is the cleanup function for a
yield dependency executed?A It is executed after the response is sent to the client. If the endpoint throws an exception, the cleanup function is still executed (similar to
try/finally).Q How deep can dependencies be nested?
A There is no strict limit, but nesting beyond three levels makes debugging more difficult. PriceTracker's three-level DI chain (get_db → get_current_user → require_admin) is the recommended upper limit.
Q Can synchronous and asynchronous dependencies be used together?
A Yes. FastAPI automatically handles synchronous and asynchronous dependencies. Synchronous dependencies run in the thread pool, while asynchronous dependencies run in the event loop.
Q Does
dependency_overrides affect other tests?A Yes, because it modifies the
app object. Be sure to call app.dependency_overrides.clear() or use fixtures to manage it after the test is complete.Q How are init arguments injected for class dependencies?
A Just like with function dependencies, init
📖 Summary
Depends()Extract repetitive logic into reusable helper functions or classes; endpoints only need to declare them and do not need to concern themselves with the implementation.- Automatic recursive resolution of sub-dependencies; caching within the same request to avoid duplicate executions
yieldImplements request-level lifecycle management for dependencies to ensure that resource creation and cleanup are paired- Routing group dependencies (
APIRouter(dependencies=[...])) allow a group of endpoints to share authentication and validation logic app.dependency_overridesis a powerful testing tool that replaces the actual dependencies of all endpoints with a single line of code
📝 Exercises
- Basic Problem (Difficulty ⭐): Create a helper function
get_dbthat returns a dictionary simulating a database connection, and inject and use it in both endpoints. Hint:db: dict = Depends(get_db) - Advanced Problem (Difficulty ⭐⭐): Implement two levels of nested dependencies:
get_db→get_current_user(read the Authorization header from the request), and inject the current user's information into the/meendpoint. Hint:user: dict = Depends(get_current_user) - Challenge (Difficulty ⭐⭐⭐): Implement a three-layer DI chain for PriceTracker:
get_db(yield + cleanup) →get_current_user(verify token) →require_subscription("pro")(check subscription level), and replace the DB dependency withdependency_overridesduring testing. Hint:yielddependency +app.dependency_overrides[get_db] = mock_fn
---|
