0
# Dependency Injection
1

2
FastAPI's dependency injection system provides automatic resolution of dependencies with hierarchical composition, caching, and security integration. Dependencies can be functions, classes, or any callable that FastAPI can inspect and resolve automatically.
3

4
## Capabilities
5

6
### Depends Function
7

8
Core function for declaring dependencies with automatic resolution and caching.
9

10
```python { .api }
11
def Depends(dependency: Optional[Callable[..., Any]] = None, *, use_cache: bool = True) -> Any:
12
    """
13
    Declare a dependency for automatic injection and resolution.
14

15
    Parameters:
16
    - dependency: Callable to be invoked as a dependency (function, class, or callable)
17
                 If None, uses the type annotation of the parameter as the dependency
18
    - use_cache: Whether to cache the result of this dependency per request
19
                Can be set to False to call the dependency function multiple times per request
20

21
    Returns:
22
    Dependency marker that FastAPI uses for automatic resolution
23

24
    Behaviors:
25
    - Dependencies are resolved recursively (dependencies can have their own dependencies)  
26
    - Results are cached per request by default to avoid multiple calls
27
    - Dependencies are resolved before calling the endpoint function
28
    - Failed dependencies prevent endpoint execution and return appropriate HTTP errors
29
    - Sub-dependencies are resolved in dependency order
30
    """
31
```
32

33
### Security Function
34

35
Function for declaring security dependencies with OAuth2 scope support and automatic OpenAPI security schema generation.
36

37
```python { .api }
38
def Security(
39
    dependency: Optional[Callable[..., Any]] = None,
40
    *,
41
    scopes: Optional[Sequence[str]] = None,
42
    use_cache: bool = True
43
) -> Any:
44
    """
45
    Declare a security dependency with optional OAuth2 scopes.
46

47
    Parameters:
48
    - dependency: Security callable (typically an OAuth2 or API key scheme)
49
    - scopes: List of OAuth2 scopes required for this endpoint
50
    - use_cache: Whether to cache the result of this dependency per request
51

52
    Returns:
53
    Security dependency marker for automatic resolution
54

55
    Behaviors:
56
    - Integrates with OpenAPI security schemas
57
    - Supports OAuth2 scopes for fine-grained permission control
58
    - Automatically generates OpenAPI security requirements
59
    - Can be combined with regular Depends() for complex security patterns
60
    """
61
```
62

63
## Dependency Patterns
64

65
### Function Dependencies
66

67
Simple functions that return values or perform setup operations.
68

69
```python { .api }
70
def get_database_connection():
71
    """
72
    Example function dependency that provides database connection.
73
    
74
    Returns:
75
    Database connection object
76
    """
77

78
def get_current_user(token: str = Depends(get_auth_token)):
79
    """
80
    Example function dependency with sub-dependency.
81
    
82
    Parameters:
83
    - token: Authentication token from sub-dependency
84
    
85
    Returns:
86
    Current user object
87
    """
88
```
89

90
### Class Dependencies
91

92
Classes used as dependencies, typically for configuration or service objects.
93

94
```python { .api }
95
class DatabaseManager:
96
    def __init__(self, connection_string: str = Depends(get_connection_string)):
97
        """
98
        Example class dependency with constructor injection.
99
        
100
        Parameters:
101
        - connection_string: Database connection string from dependency
102
        """
103
        
104
    def get_connection(self):
105
        """Get database connection."""
106
```
107

108
### Async Dependencies
109

110
Dependencies that perform asynchronous operations.
111

112
```python { .api }
113
async def get_async_resource():
114
    """
115
    Example async dependency for I/O operations.
116
    
117
    Returns:
118
    Async resource or data
119
    """
120

121
async def validate_async_token(token: str = Depends(get_token)):
122
    """
123
    Example async dependency with token validation.
124
    
125
    Parameters:
126
    - token: Token to validate
127
    
128
    Returns:
129
    Validated user information
130
    
131
    Raises:
132
    HTTPException: If token is invalid
133
    """
134
```
135

136
### Generator Dependencies
137

138
Dependencies that provide resources with automatic cleanup using context managers.
139

140
```python { .api }
141
def get_database_session():
142
    """
143
    Example generator dependency with automatic cleanup.
144
    
145
    Yields:
146
    Database session object
147
    
148
    Note: Cleanup code runs after endpoint execution
149
    """
150

151
async def get_async_session():
152
    """
153
    Example async generator dependency with automatic cleanup.
154
    
155
    Yields:
156
    Async database session object
157
    """
158
```
159

160
## Usage Examples
161

162
### Basic Dependencies
163

164
```python
165
from fastapi import FastAPI, Depends, HTTPException
166

167
app = FastAPI()
168

169
# Simple dependency function
170
def get_database():
171
    # In real app, this would return actual database connection
172
    return {"type": "database", "status": "connected"}
173

174
def get_current_user(db = Depends(get_database)):
175
    # Simulate user lookup
176
    user = db.get("current_user")  # This would be actual DB query
177
    if not user:
178
        raise HTTPException(status_code=401, detail="User not found")
179
    return {"user_id": 1, "username": "john", "is_active": True}
180

181
@app.get("/users/me")
182
async def get_user_profile(current_user: dict = Depends(get_current_user)):
183
    return current_user
184

185
@app.get("/items/")
186
async def get_items(
187
    db = Depends(get_database),
188
    current_user: dict = Depends(get_current_user)
189
):
190
    # Both dependencies are resolved automatically
191
    return {
192
        "items": ["item1", "item2"],
193
        "user": current_user["username"],
194
        "db_status": db["status"]
195
    }
196
```
197

198
### Class-Based Dependencies
199

200
```python
201
from fastapi import FastAPI, Depends
202

203
app = FastAPI()
204

205
class Settings:
206
    def __init__(self):
207
        self.api_key = "secret-key"
208
        self.debug = True
209
        self.max_connections = 100
210

211
class DatabaseManager:
212
    def __init__(self, settings: Settings = Depends(Settings)):
213
        self.settings = settings
214
        self.connection_pool = f"pool-{settings.max_connections}"
215
    
216
    def get_connection(self):
217
        return f"connection-{self.settings.api_key}"
218

219
@app.get("/config")
220
async def get_config(
221
    settings: Settings = Depends(Settings),
222
    db: DatabaseManager = Depends(DatabaseManager)
223
):
224
    return {
225
        "debug": settings.debug,
226
        "connection": db.get_connection()
227
    }
228
```
229

230
### Security Dependencies
231

232
```python
233
from fastapi import FastAPI, Depends, HTTPException, Security
234
from fastapi.security import HTTPBearer, OAuth2PasswordBearer
235

236
app = FastAPI()
237

238
# HTTP Bearer token security
239
security = HTTPBearer()
240

241
def verify_token(token: str = Depends(security)):
242
    if token.credentials != "valid-token":
243
        raise HTTPException(status_code=401, detail="Invalid token")
244
    return token.credentials
245

246
# OAuth2 with scopes
247
oauth2_scheme = OAuth2PasswordBearer(
248
    tokenUrl="token",
249
    scopes={"read": "Read access", "write": "Write access"}
250
)
251

252
def get_current_user(token: str = Depends(oauth2_scheme)):
253
    # Verify token and return user
254
    return {"user_id": 1, "username": "john", "scopes": ["read", "write"]}
255

256
def require_scopes(required_scopes: list):
257
    def check_scopes(current_user: dict = Security(get_current_user, scopes=required_scopes)):
258
        user_scopes = current_user.get("scopes", [])
259
        for scope in required_scopes:
260
            if scope not in user_scopes:
261
                raise HTTPException(status_code=403, detail="Insufficient permissions")
262
        return current_user
263
    return check_scopes
264

265
@app.get("/public")
266
async def public_endpoint():
267
    return {"message": "This is public"}
268

269
@app.get("/protected")
270
async def protected_endpoint(user: dict = Depends(get_current_user)):
271
    return {"message": f"Hello {user['username']}"}
272

273
@app.post("/admin/users")
274
async def create_user(user: dict = Security(get_current_user, scopes=["write"])):
275
    return {"message": "User created", "by": user["username"]}
276
```
277

278
### Async Dependencies
279

280
```python
281
import asyncio
282
from fastapi import FastAPI, Depends, HTTPException
283

284
app = FastAPI()
285

286
async def get_async_database():
287
    # Simulate async database connection
288
    await asyncio.sleep(0.1)
289
    return {"type": "async_db", "status": "connected"}
290

291
async def get_external_data():
292
    # Simulate external API call
293
    await asyncio.sleep(0.2)
294
    return {"data": "external_value", "timestamp": "2023-01-01"}
295

296
async def validate_user_async(
297
    db = Depends(get_async_database),
298
    external_data = Depends(get_external_data)
299
):
300
    # Perform async validation
301
    if external_data["data"] != "external_value":
302
        raise HTTPException(status_code=400, detail="Validation failed")
303
    
304
    return {
305
        "user_id": 1,
306
        "username": "async_user",
307
        "validated_at": external_data["timestamp"]
308
    }
309

310
@app.get("/async-endpoint")
311
async def async_endpoint(
312
    user = Depends(validate_user_async),
313
    db = Depends(get_async_database)
314
):
315
    return {
316
        "user": user,
317
        "db_status": db["status"]
318
    }
319
```
320

321
### Generator Dependencies with Cleanup
322

323
```python
324
from fastapi import FastAPI, Depends
325
import contextlib
326

327
app = FastAPI()
328

329
def get_database_session():
330
    """Generator dependency with automatic cleanup."""
331
    print("Creating database session")
332
    session = {"id": "session_123", "active": True}
333
    try:
334
        yield session
335
    finally:
336
        print("Closing database session")
337
        session["active"] = False
338

339
async def get_async_resource():
340
    """Async generator dependency with cleanup."""
341
    print("Acquiring async resource")
342
    resource = {"handle": "resource_456", "open": True}
343
    try:
344
        yield resource
345
    finally:
346
        print("Releasing async resource")
347
        resource["open"] = False
348

349
@app.get("/with-cleanup")
350
async def endpoint_with_cleanup(
351
    db_session = Depends(get_database_session),
352
    async_resource = Depends(get_async_resource)
353
):
354
    # Resources are automatically cleaned up after response
355
    return {
356
        "session_id": db_session["id"],
357
        "resource_handle": async_resource["handle"]
358
    }
359
```
360

361
### Dependency Overriding
362

363
```python
364
from fastapi import FastAPI, Depends
365
from fastapi.testclient import TestClient
366

367
app = FastAPI()
368

369
def get_database():
370
    return {"type": "production", "host": "prod.db"}
371

372
def get_current_user(db = Depends(get_database)):
373
    return {"user_id": 1, "username": "prod_user"}
374

375
@app.get("/data")
376
async def get_data(user = Depends(get_current_user)):
377
    return {"data": "sensitive_data", "user": user["username"]}
378

379
# Test with dependency overrides
380
def get_test_database():
381
    return {"type": "test", "host": "test.db"}
382

383
def get_test_user(db = Depends(get_test_database)):
384
    return {"user_id": 999, "username": "test_user"}
385

386
# In tests:
387
# app.dependency_overrides[get_database] = get_test_database
388
# app.dependency_overrides[get_current_user] = get_test_user
389
```
390

391
### Sub-Dependencies and Caching
392

393
```python
394
from fastapi import FastAPI, Depends
395
import time
396

397
app = FastAPI()
398

399
def expensive_computation():
400
    """Expensive operation that should be cached."""
401
    print("Performing expensive computation...")
402
    time.sleep(1)  # Simulate expensive operation
403
    return {"result": "computed_value", "timestamp": time.time()}
404

405
def get_user_data(computation = Depends(expensive_computation)):
406
    return {
407
        "user_id": 1,
408
        "computed_result": computation["result"]
409
    }
410

411
def get_admin_data(computation = Depends(expensive_computation)):
412
    return {
413
        "admin_id": 1,
414
        "computed_result": computation["result"]
415
    }
416

417
@app.get("/user-and-admin")
418
async def get_both_data(
419
    user_data = Depends(get_user_data),
420
    admin_data = Depends(get_admin_data)
421
):
422
    # expensive_computation() is called only once due to caching
423
    return {
424
        "user": user_data,
425
        "admin": admin_data
426
    }
427

428
# Disable caching for specific dependency
429
def no_cache_computation():
430
    print("Computation without caching...")
431
    return {"result": "fresh_value", "timestamp": time.time()}
432

433
def get_fresh_data(computation = Depends(no_cache_computation, use_cache=False)):
434
    return {"fresh_result": computation["result"]}
435

436
@app.get("/fresh-data")
437
async def get_multiple_fresh(
438
    data1 = Depends(get_fresh_data),
439
    data2 = Depends(get_fresh_data)
440
):
441
    # no_cache_computation() is called twice due to use_cache=False
442
    return {"data1": data1, "data2": data2}
443
```