pypi-fastapi

1
# Dependency Injection
2

3
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.
4

5
## Capabilities
6

7
### Depends Function
8

9
Core dependency injection function that declares dependencies with optional caching control.
10

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

16
    Parameters:
17
    - dependency: Callable that provides the dependency value
18
    - use_cache: Whether to cache the dependency result within a request
19

20
    Returns:
21
    Dependency declaration for use in function signatures
22

23
    The dependency callable can be:
24
    - A function that returns a value
25
    - A class constructor
26
    - Another dependency that uses Depends()
27
    - A callable with its own dependencies
28
    """
29
```
30

31
### Security Dependencies
32

33
Special dependency function for security-related dependencies with scope support for authorization.
34

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

45
    Parameters:
46
    - dependency: Security scheme callable (e.g., OAuth2PasswordBearer)
47
    - scopes: List of required OAuth2 scopes for authorization
48
    - use_cache: Whether to cache the dependency result within a request
49

50
    Returns:
51
    Security dependency declaration for use in function signatures
52

53
    Used for endpoints that require specific permissions or scopes.
54
    """
55
```
56

57
## Usage Examples
58

59
### Basic Dependencies
60

61
```python
62
from fastapi import FastAPI, Depends
63

64
app = FastAPI()
65

66
# Simple dependency function
67
def get_db():
68
    db = {"connection": "postgresql://..."}
69
    try:
70
        yield db
71
    finally:
72
        # Close connection
73
        pass
74

75
def get_current_user():
76
    return {"user_id": 1, "username": "john"}
77

78
@app.get("/items/")
79
def read_items(
80
    db=Depends(get_db),
81
    current_user=Depends(get_current_user)
82
):
83
    return {"db": db, "user": current_user}
84
```
85

86
### Class-based Dependencies
87

88
```python
89
from fastapi import FastAPI, Depends
90

91
app = FastAPI()
92

93
class DatabaseService:
94
    def __init__(self):
95
        self.connection = "postgresql://..."
96
    
97
    def get_connection(self):
98
        return self.connection
99

100
class UserService:
101
    def __init__(self, db: DatabaseService = Depends(DatabaseService)):
102
        self.db = db
103
    
104
    def get_current_user(self):
105
        return {"user_id": 1, "username": "john"}
106

107
@app.get("/users/me")
108
def get_user_profile(user_service: UserService = Depends(UserService)):
109
    return user_service.get_current_user()
110
```
111

112
### Dependency with Parameters
113

114
```python
115
from fastapi import FastAPI, Depends, Query
116

117
app = FastAPI()
118

119
def common_parameters(
120
    skip: int = Query(0, ge=0),
121
    limit: int = Query(10, ge=1, le=100)
122
):
123
    return {"skip": skip, "limit": limit}
124

125
@app.get("/items/")
126
def read_items(commons: dict = Depends(common_parameters)):
127
    return {"params": commons}
128

129
@app.get("/users/")
130
def read_users(commons: dict = Depends(common_parameters)):
131
    return {"params": commons}
132
```
133

134
### Nested Dependencies
135

136
```python
137
from fastapi import FastAPI, Depends, HTTPException
138

139
app = FastAPI()
140

141
def get_db():
142
    return {"connection": "active"}
143

144
def get_user_service(db=Depends(get_db)):
145
    return {"db": db, "service": "user_service"}
146

147
def get_current_user(user_service=Depends(get_user_service)):
148
    # Authentication logic using user_service
149
    return {"user_id": 1, "username": "john"}
150

151
def get_admin_user(current_user=Depends(get_current_user)):
152
    if not current_user.get("is_admin"):
153
        raise HTTPException(status_code=403, detail="Admin required")
154
    return current_user
155

156
@app.get("/admin/users")
157
def list_users(admin_user=Depends(get_admin_user)):
158
    return {"message": "Admin access granted", "admin": admin_user}
159
```
160

161
### Dependency Caching
162

163
```python
164
from fastapi import FastAPI, Depends
165
import time
166

167
app = FastAPI()
168

169
# Expensive operation that should be cached
170
def get_expensive_data():
171
    print("Computing expensive data...")
172
    time.sleep(1)  # Simulate expensive operation
173
    return {"data": "expensive_result", "timestamp": time.time()}
174

175
# Cached dependency (default behavior)
176
def cached_dependency():
177
    return get_expensive_data()
178

179
# Non-cached dependency
180
def non_cached_dependency():
181
    return get_expensive_data()
182

183
@app.get("/cached")
184
def endpoint_with_cached_deps(
185
    data1=Depends(cached_dependency),
186
    data2=Depends(cached_dependency)  # Same result, computed only once
187
):
188
    return {"data1": data1, "data2": data2}
189

190
@app.get("/non-cached")
191
def endpoint_with_non_cached_deps(
192
    data1=Depends(non_cached_dependency, use_cache=False),
193
    data2=Depends(non_cached_dependency, use_cache=False)  # Computed twice
194
):
195
    return {"data1": data1, "data2": data2}
196
```
197

198
### Global Dependencies
199

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

203
# Global authentication dependency
204
def verify_api_key(api_key: str = Header(...)):
205
    if api_key != "secret-api-key":
206
        raise HTTPException(status_code=401, detail="Invalid API key")
207
    return api_key
208

209
# Apply dependency to entire application
210
app = FastAPI(dependencies=[Depends(verify_api_key)])
211

212
@app.get("/protected-endpoint")
213
def protected_endpoint():
214
    return {"message": "This endpoint requires API key"}
215

216
@app.get("/another-protected-endpoint")
217
def another_protected_endpoint():
218
    return {"message": "This endpoint also requires API key"}
219
```
220

221
### Router-level Dependencies
222

223
```python
224
from fastapi import FastAPI, APIRouter, Depends, HTTPException
225

226
app = FastAPI()
227

228
def admin_required():
229
    # Authentication logic
230
    return {"role": "admin"}
231

232
# Router with shared dependencies
233
admin_router = APIRouter(
234
    prefix="/admin",
235
    dependencies=[Depends(admin_required)]
236
)
237

238
@admin_router.get("/users")
239
def admin_list_users():
240
    return {"users": ["admin", "user1", "user2"]}
241

242
@admin_router.delete("/users/{user_id}")
243
def admin_delete_user(user_id: int):
244
    return {"message": f"User {user_id} deleted"}
245

246
app.include_router(admin_router)
247
```
248

249
### Security Dependencies with Scopes
250

251
```python
252
from fastapi import FastAPI, Depends, HTTPException, Security
253
from fastapi.security import OAuth2PasswordBearer
254
from typing import List
255

256
app = FastAPI()
257

258
oauth2_scheme = OAuth2PasswordBearer(
259
    tokenUrl="token",
260
    scopes={
261
        "read": "Read access",
262
        "write": "Write access",
263
        "admin": "Admin access"
264
    }
265
)
266

267
def get_current_user(token: str = Depends(oauth2_scheme)):
268
    # Decode token and return user
269
    return {"username": "john", "scopes": ["read", "write"]}
270

271
def check_scopes(required_scopes: List[str]):
272
    def scopes_checker(
273
        current_user=Security(oauth2_scheme, scopes=required_scopes)
274
    ):
275
        user_scopes = current_user.get("scopes", [])
276
        for scope in required_scopes:
277
            if scope not in user_scopes:
278
                raise HTTPException(
279
                    status_code=403,
280
                    detail=f"Not enough permissions. Required: {required_scopes}"
281
                )
282
        return current_user
283
    return scopes_checker
284

285
@app.get("/read-data")
286
def read_data(
287
    current_user=Security(oauth2_scheme, scopes=["read"])
288
):
289
    return {"data": "sensitive data", "user": current_user}
290

291
@app.post("/write-data")
292
def write_data(
293
    current_user=Security(oauth2_scheme, scopes=["write"])
294
):
295
    return {"message": "Data written", "user": current_user}
296

297
@app.delete("/admin-action")
298
def admin_action(
299
    current_user=Security(oauth2_scheme, scopes=["admin"])
300
):
301
    return {"message": "Admin action performed", "user": current_user}
302
```
303

304
### Dependency Override for Testing
305

306
```python
307
from fastapi import FastAPI, Depends
308
from fastapi.testclient import TestClient
309

310
app = FastAPI()
311

312
def get_db():
313
    return {"connection": "production_db"}
314

315
def get_current_user():
316
    return {"user_id": 1, "username": "john"}
317

318
@app.get("/items/")
319
def read_items(
320
    db=Depends(get_db),
321
    current_user=Depends(get_current_user)
322
):
323
    return {"db": db, "user": current_user}
324

325
# Test with dependency overrides
326
def test_read_items():
327
    def override_get_db():
328
        return {"connection": "test_db"}
329
    
330
    def override_get_current_user():
331
        return {"user_id": 999, "username": "test_user"}
332
    
333
    app.dependency_overrides[get_db] = override_get_db
334
    app.dependency_overrides[get_current_user] = override_get_current_user
335
    
336
    client = TestClient(app)
337
    response = client.get("/items/")
338
    
339
    # Clean up overrides
340
    app.dependency_overrides = {}
341
    
342
    assert response.status_code == 200
343
    data = response.json()
344
    assert data["db"]["connection"] == "test_db"
345
    assert data["user"]["username"] == "test_user"
346
```
347

348
### Dependency with Cleanup
349

350
```python
351
from fastapi import FastAPI, Depends
352
import asyncio
353

354
app = FastAPI()
355

356
class DatabaseConnection:
357
    def __init__(self):
358
        self.connection = None
359
    
360
    async def connect(self):
361
        print("Connecting to database...")
362
        self.connection = "active_connection"
363
        return self
364
    
365
    async def disconnect(self):
366
        print("Disconnecting from database...")
367
        self.connection = None
368

369
async def get_database():
370
    db = DatabaseConnection()
371
    await db.connect()
372
    try:
373
        yield db
374
    finally:
375
        await db.disconnect()
376

377
@app.get("/items/")
378
async def read_items(db: DatabaseConnection = Depends(get_database)):
379
    return {"connection_status": db.connection}
380
```