0
# Dependency Injection
1
2
FastAPI provides a powerful dependency injection system that allows sharing code, database connections, authentication, and other common functionality across endpoints. Dependencies are cached by default and can be overridden for testing.
3
4
## Capabilities
5
6
### Depends Function
7
8
Core dependency injection function that declares dependencies with optional caching control.
9
10
```python { .api }
11
def Depends(dependency: Callable = None, *, use_cache: bool = True) -> Any:
12
"""
13
Declare a dependency for dependency injection.
14
15
Parameters:
16
- dependency: Callable that provides the dependency value
17
- use_cache: Whether to cache the dependency result within a request
18
19
Returns:
20
Dependency declaration for use in function signatures
21
22
The dependency callable can be:
23
- A function that returns a value
24
- A class constructor
25
- Another dependency that uses Depends()
26
- A callable with its own dependencies
27
"""
28
```
29
30
### Security Dependencies
31
32
Special dependency function for security-related dependencies with scope support for authorization.
33
34
```python { .api }
35
def Security(
36
dependency: Callable = None,
37
*,
38
scopes: List[str] = None,
39
use_cache: bool = True,
40
) -> Any:
41
"""
42
Declare a security dependency with OAuth2 scopes support.
43
44
Parameters:
45
- dependency: Security scheme callable (e.g., OAuth2PasswordBearer)
46
- scopes: List of required OAuth2 scopes for authorization
47
- use_cache: Whether to cache the dependency result within a request
48
49
Returns:
50
Security dependency declaration for use in function signatures
51
52
Used for endpoints that require specific permissions or scopes.
53
"""
54
```
55
56
## Usage Examples
57
58
### Basic Dependencies
59
60
```python
61
from fastapi import FastAPI, Depends
62
63
app = FastAPI()
64
65
# Simple dependency function
66
def get_db():
67
db = {"connection": "postgresql://..."}
68
try:
69
yield db
70
finally:
71
# Close connection
72
pass
73
74
def get_current_user():
75
return {"user_id": 1, "username": "john"}
76
77
@app.get("/items/")
78
def read_items(
79
db=Depends(get_db),
80
current_user=Depends(get_current_user)
81
):
82
return {"db": db, "user": current_user}
83
```
84
85
### Class-based Dependencies
86
87
```python
88
from fastapi import FastAPI, Depends
89
90
app = FastAPI()
91
92
class DatabaseService:
93
def __init__(self):
94
self.connection = "postgresql://..."
95
96
def get_connection(self):
97
return self.connection
98
99
class UserService:
100
def __init__(self, db: DatabaseService = Depends(DatabaseService)):
101
self.db = db
102
103
def get_current_user(self):
104
return {"user_id": 1, "username": "john"}
105
106
@app.get("/users/me")
107
def get_user_profile(user_service: UserService = Depends(UserService)):
108
return user_service.get_current_user()
109
```
110
111
### Dependency with Parameters
112
113
```python
114
from fastapi import FastAPI, Depends, Query
115
116
app = FastAPI()
117
118
def common_parameters(
119
skip: int = Query(0, ge=0),
120
limit: int = Query(10, ge=1, le=100)
121
):
122
return {"skip": skip, "limit": limit}
123
124
@app.get("/items/")
125
def read_items(commons: dict = Depends(common_parameters)):
126
return {"params": commons}
127
128
@app.get("/users/")
129
def read_users(commons: dict = Depends(common_parameters)):
130
return {"params": commons}
131
```
132
133
### Nested Dependencies
134
135
```python
136
from fastapi import FastAPI, Depends, HTTPException
137
138
app = FastAPI()
139
140
def get_db():
141
return {"connection": "active"}
142
143
def get_user_service(db=Depends(get_db)):
144
return {"db": db, "service": "user_service"}
145
146
def get_current_user(user_service=Depends(get_user_service)):
147
# Authentication logic using user_service
148
return {"user_id": 1, "username": "john"}
149
150
def get_admin_user(current_user=Depends(get_current_user)):
151
if not current_user.get("is_admin"):
152
raise HTTPException(status_code=403, detail="Admin required")
153
return current_user
154
155
@app.get("/admin/users")
156
def list_users(admin_user=Depends(get_admin_user)):
157
return {"message": "Admin access granted", "admin": admin_user}
158
```
159
160
### Dependency Caching
161
162
```python
163
from fastapi import FastAPI, Depends
164
import time
165
166
app = FastAPI()
167
168
# Expensive operation that should be cached
169
def get_expensive_data():
170
print("Computing expensive data...")
171
time.sleep(1) # Simulate expensive operation
172
return {"data": "expensive_result", "timestamp": time.time()}
173
174
# Cached dependency (default behavior)
175
def cached_dependency():
176
return get_expensive_data()
177
178
# Non-cached dependency
179
def non_cached_dependency():
180
return get_expensive_data()
181
182
@app.get("/cached")
183
def endpoint_with_cached_deps(
184
data1=Depends(cached_dependency),
185
data2=Depends(cached_dependency) # Same result, computed only once
186
):
187
return {"data1": data1, "data2": data2}
188
189
@app.get("/non-cached")
190
def endpoint_with_non_cached_deps(
191
data1=Depends(non_cached_dependency, use_cache=False),
192
data2=Depends(non_cached_dependency, use_cache=False) # Computed twice
193
):
194
return {"data1": data1, "data2": data2}
195
```
196
197
### Global Dependencies
198
199
```python
200
from fastapi import FastAPI, Depends, HTTPException
201
202
# Global authentication dependency
203
def verify_api_key(api_key: str = Header(...)):
204
if api_key != "secret-api-key":
205
raise HTTPException(status_code=401, detail="Invalid API key")
206
return api_key
207
208
# Apply dependency to entire application
209
app = FastAPI(dependencies=[Depends(verify_api_key)])
210
211
@app.get("/protected-endpoint")
212
def protected_endpoint():
213
return {"message": "This endpoint requires API key"}
214
215
@app.get("/another-protected-endpoint")
216
def another_protected_endpoint():
217
return {"message": "This endpoint also requires API key"}
218
```
219
220
### Router-level Dependencies
221
222
```python
223
from fastapi import FastAPI, APIRouter, Depends, HTTPException
224
225
app = FastAPI()
226
227
def admin_required():
228
# Authentication logic
229
return {"role": "admin"}
230
231
# Router with shared dependencies
232
admin_router = APIRouter(
233
prefix="/admin",
234
dependencies=[Depends(admin_required)]
235
)
236
237
@admin_router.get("/users")
238
def admin_list_users():
239
return {"users": ["admin", "user1", "user2"]}
240
241
@admin_router.delete("/users/{user_id}")
242
def admin_delete_user(user_id: int):
243
return {"message": f"User {user_id} deleted"}
244
245
app.include_router(admin_router)
246
```
247
248
### Security Dependencies with Scopes
249
250
```python
251
from fastapi import FastAPI, Depends, HTTPException, Security
252
from fastapi.security import OAuth2PasswordBearer
253
from typing import List
254
255
app = FastAPI()
256
257
oauth2_scheme = OAuth2PasswordBearer(
258
tokenUrl="token",
259
scopes={
260
"read": "Read access",
261
"write": "Write access",
262
"admin": "Admin access"
263
}
264
)
265
266
def get_current_user(token: str = Depends(oauth2_scheme)):
267
# Decode token and return user
268
return {"username": "john", "scopes": ["read", "write"]}
269
270
def check_scopes(required_scopes: List[str]):
271
def scopes_checker(
272
current_user=Security(oauth2_scheme, scopes=required_scopes)
273
):
274
user_scopes = current_user.get("scopes", [])
275
for scope in required_scopes:
276
if scope not in user_scopes:
277
raise HTTPException(
278
status_code=403,
279
detail=f"Not enough permissions. Required: {required_scopes}"
280
)
281
return current_user
282
return scopes_checker
283
284
@app.get("/read-data")
285
def read_data(
286
current_user=Security(oauth2_scheme, scopes=["read"])
287
):
288
return {"data": "sensitive data", "user": current_user}
289
290
@app.post("/write-data")
291
def write_data(
292
current_user=Security(oauth2_scheme, scopes=["write"])
293
):
294
return {"message": "Data written", "user": current_user}
295
296
@app.delete("/admin-action")
297
def admin_action(
298
current_user=Security(oauth2_scheme, scopes=["admin"])
299
):
300
return {"message": "Admin action performed", "user": current_user}
301
```
302
303
### Dependency Override for Testing
304
305
```python
306
from fastapi import FastAPI, Depends
307
from fastapi.testclient import TestClient
308
309
app = FastAPI()
310
311
def get_db():
312
return {"connection": "production_db"}
313
314
def get_current_user():
315
return {"user_id": 1, "username": "john"}
316
317
@app.get("/items/")
318
def read_items(
319
db=Depends(get_db),
320
current_user=Depends(get_current_user)
321
):
322
return {"db": db, "user": current_user}
323
324
# Test with dependency overrides
325
def test_read_items():
326
def override_get_db():
327
return {"connection": "test_db"}
328
329
def override_get_current_user():
330
return {"user_id": 999, "username": "test_user"}
331
332
app.dependency_overrides[get_db] = override_get_db
333
app.dependency_overrides[get_current_user] = override_get_current_user
334
335
client = TestClient(app)
336
response = client.get("/items/")
337
338
# Clean up overrides
339
app.dependency_overrides = {}
340
341
assert response.status_code == 200
342
data = response.json()
343
assert data["db"]["connection"] == "test_db"
344
assert data["user"]["username"] == "test_user"
345
```
346
347
### Dependency with Cleanup
348
349
```python
350
from fastapi import FastAPI, Depends
351
import asyncio
352
353
app = FastAPI()
354
355
class DatabaseConnection:
356
def __init__(self):
357
self.connection = None
358
359
async def connect(self):
360
print("Connecting to database...")
361
self.connection = "active_connection"
362
return self
363
364
async def disconnect(self):
365
print("Disconnecting from database...")
366
self.connection = None
367
368
async def get_database():
369
db = DatabaseConnection()
370
await db.connect()
371
try:
372
yield db
373
finally:
374
await db.disconnect()
375
376
@app.get("/items/")
377
async def read_items(db: DatabaseConnection = Depends(get_database)):
378
return {"connection_status": db.connection}
379
```