Implement FastAPI dependency injection system for shared logic, database connections, authentication, and code reuse across path operations.
Master FastAPI's powerful dependency injection system to share logic, manage database connections, enforce security, and minimize code repetition across your API endpoints.
This skill teaches you to implement FastAPI's dependency injection pattern using the `Depends()` function. You'll learn to create reusable "dependable" functions that can be injected into path operations, enabling shared logic for authentication, database connections, common parameters, and more.
Dependency Injection in FastAPI allows path operation functions to declare what they need to work. FastAPI automatically provides (injects) those dependencies.
**Key benefits:**
Define a function that accepts the same parameters as a path operation function:
```python
from typing import Annotated
from fastapi import Depends, FastAPI
app = FastAPI()
async def common_parameters(
q: str | None = None,
skip: int = 0,
limit: int = 100
):
return {"q": q, "skip": skip, "limit": limit}
```
**Key points:**
Use `Depends()` with `Annotated` to declare dependencies:
```python
@app.get("/items/")
async def read_items(
commons: Annotated[dict, Depends(common_parameters)]
):
return commons
@app.get("/users/")
async def read_users(
commons: Annotated[dict, Depends(common_parameters)]
):
return commons
```
**Pattern:** `Annotated[ReturnType, Depends(dependency_function)]`
Share database sessions across endpoints:
```python
from sqlalchemy.orm import Session
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
@app.get("/users/")
def read_users(
db: Annotated[Session, Depends(get_db)],
skip: int = 0,
limit: int = 100
):
users = db.query(User).offset(skip).limit(limit).all()
return users
```
Create reusable authentication logic:
```python
from fastapi import HTTPException, status
from fastapi.security import OAuth2PasswordBearer
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
async def get_current_user(
token: Annotated[str, Depends(oauth2_scheme)]
):
user = verify_token(token)
if not user:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid authentication credentials"
)
return user
@app.get("/users/me")
async def read_users_me(
current_user: Annotated[User, Depends(get_current_user)]
):
return current_user
```
Dependencies can be classes with `__call__` methods:
```python
class CommonQueryParams:
def __init__(
self,
q: str | None = None,
skip: int = 0,
limit: int = 100
):
self.q = q
self.skip = skip
self.limit = limit
@app.get("/items/")
async def read_items(
commons: Annotated[CommonQueryParams, Depends()]
):
return commons
```
**Shortcut:** When the dependency is a class, you can omit the function call: `Depends()` instead of `Depends(CommonQueryParams)`
Dependencies can depend on other dependencies:
```python
async def verify_token(token: Annotated[str, Depends(oauth2_scheme)]):
return decode_token(token)
async def get_current_user(
payload: Annotated[dict, Depends(verify_token)]
):
user_id = payload.get("sub")
return get_user(user_id)
async def get_current_active_user(
current_user: Annotated[User, Depends(get_current_user)]
):
if not current_user.is_active:
raise HTTPException(status_code=400, detail="Inactive user")
return current_user
```
Apply dependencies to all path operations in an app or router:
```python
app = FastAPI(dependencies=[Depends(verify_api_key)])
router = APIRouter(
prefix="/admin",
dependencies=[Depends(verify_admin)]
)
```
For setup and teardown logic (like database connections):
```python
async def get_db():
db = Database()
await db.connect()
try:
yield db
finally:
await db.disconnect()
@app.get("/items/")
async def read_items(db: Annotated[Database, Depends(get_db)]):
return await db.query("SELECT * FROM items")
```
**Execution order:**
1. Code before `yield` runs before the path operation
2. The yielded value is injected
3. Code after `yield` runs after the path operation completes
Raise HTTPExceptions in dependencies for proper error responses:
```python
async def verify_api_key(api_key: Annotated[str, Header()]):
if api_key != "secret-key":
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Invalid API Key"
)
return api_key
```
```python
async def pagination_params(skip: int = 0, limit: int = 100):
return {"skip": skip, "limit": limit}
```
```python
async def rate_limiter(request: Request):
# Check rate limit logic
if exceeded_rate_limit(request.client.host):
raise HTTPException(status_code=429, detail="Too many requests")
```
```python
async def require_admin(current_user: Annotated[User, Depends(get_current_user)]):
if not current_user.is_admin:
raise HTTPException(status_code=403, detail="Admin access required")
return current_user
```
1. **Use `Annotated`**: Always use `Annotated[Type, Depends(...)]` for type safety and editor support
2. **Keep dependencies focused**: Each dependency should do one thing well
3. **Use `yield` for cleanup**: Database connections, file handles, etc.
4. **Raise HTTPException**: For proper API error responses
5. **Async when needed**: Use `async def` for I/O operations, `def` for CPU-bound work
6. **Cache when appropriate**: FastAPI caches dependency results within a single request
7. **Document dependencies**: Add docstrings explaining what dependencies provide
Override dependencies in tests:
```python
from fastapi.testclient import TestClient
def override_get_db():
return mock_db
app.dependency_overrides[get_db] = override_get_db
client = TestClient(app)
```
Leave a review
No reviews yet. Be the first to review this skill!
# Download SKILL.md from killerskills.ai/api/skills/fastapi-dependency-injection/raw