FastAPI Sub-dependencies
In FastAPI applications, as they grow in complexity, you'll often need to build more sophisticated dependency structures. Sub-dependencies are a powerful feature that allows you to create dependency chains where one dependency depends on another. This modularity enhances code reusability and maintains clean separation of concerns in your application.
What Are Sub-dependencies?
Sub-dependencies are dependencies that themselves depend on other dependencies. This creates a chain or tree of dependencies where FastAPI automatically resolves each level for you. Think of them as building blocks that you can arrange and reuse in different parts of your application.
Basic Sub-dependency Example
Let's start with a simple example to understand how sub-dependencies work:
from fastapi import Depends, FastAPI
app = FastAPI()
# First level dependency
def get_query_parameter(q: str = None):
return q or "default_query"
# Second level dependency (sub-dependency)
def get_processed_query(query: str = Depends(get_query_parameter)):
return f"Processed: {query.upper()}"
@app.get("/items/")
async def read_items(processed_query: str = Depends(get_processed_query)):
return {"processed_query": processed_query}
When you access the /items/
endpoint:
- FastAPI first executes
get_query_parameter
to retrieve the query value - Then passes that result to
get_processed_query
- Finally, the result of
get_processed_query
is passed to the endpoint function
If you visit /items/?q=hello
, the response would be:
{
"processed_query": "Processed: HELLO"
}
If no query parameter is provided, it would return:
{
"processed_query": "Processed: DEFAULT_QUERY"
}
Building More Complex Dependency Chains
Let's build a more practical example using sub-dependencies for user authentication and authorization:
from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from typing import Optional, List
app = FastAPI()
# Mock database
fake_users_db = {
"john": {
"username": "john",
"full_name": "John Doe",
"email": "[email protected]",
"hashed_password": "fakehashedsecret",
"roles": ["user"]
},
"alice": {
"username": "alice",
"full_name": "Alice Smith",
"email": "[email protected]",
"hashed_password": "fakehashedsecret2",
"roles": ["user", "admin"]
}
}
# First level dependency
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
# Second level dependency (depends on oauth2_scheme)
def get_current_user(token: str = Depends(oauth2_scheme)):
user = fake_users_db.get(token)
if not user:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid authentication credentials",
headers={"WWW-Authenticate": "Bearer"},
)
return user
# Third level dependency (depends on get_current_user)
def get_admin_user(current_user: dict = Depends(get_current_user)):
if "admin" not in current_user.get("roles", []):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Insufficient permissions"
)
return current_user
# Using second level dependency
@app.get("/users/me")
async def read_users_me(current_user: dict = Depends(get_current_user)):
return current_user
# Using third level dependency
@app.get("/admin/")
async def read_admin_data(admin_user: dict = Depends(get_admin_user)):
return {"admin_data": "Top secret information", "admin_name": admin_user["username"]}
In this example:
oauth2_scheme
is our first-level dependency that extracts the tokenget_current_user
depends onoauth2_scheme
to get and validate a userget_admin_user
depends onget_current_user
to check if the user is an admin
Sub-dependencies in Classes
You can also use sub-dependencies with dependency classes. This approach is particularly useful for organizing complex dependency logic:
from fastapi import Depends, FastAPI, Header, HTTPException
from typing import Optional, List
app = FastAPI()
class QueryValidator:
def __init__(self, max_length: int = 50):
self.max_length = max_length
def __call__(self, q: Optional[str] = None):
if q and len(q) > self.max_length:
raise HTTPException(status_code=400, detail=f"Query too long (max {self.max_length} characters)")
return q or ""
class QueryProcessor:
def __init__(
self,
validator: QueryValidator = Depends(QueryValidator),
convert_to_uppercase: bool = True
):
self.validator = validator
self.convert_to_uppercase = convert_to_uppercase
def __call__(self, q: str = Depends(validator)):
# Process the already validated query
processed = q
if self.convert_to_uppercase:
processed = processed.upper()
return processed
@app.get("/search/")
async def search(
processed_query: str = Depends(QueryProcessor()),
user_agent: str = Header(None)
):
return {
"processed_query": processed_query,
"user_agent": user_agent
}
In this example:
QueryValidator
validates that the query isn't too longQueryProcessor
depends onQueryValidator
and transforms the query- Our endpoint function depends on
QueryProcessor
Dependency Overrides with Sub-dependencies
When testing your application, you might need to replace parts of your dependency chain. FastAPI allows you to override dependencies:
from fastapi import Depends, FastAPI
from fastapi.testclient import TestClient
app = FastAPI()
async def get_db():
# In a real app, this might be a database connection
return {"data": "Real database connection"}
async def get_user_service(db = Depends(get_db)):
return {"user_service": "Real service", "db": db}
@app.get("/users/")
async def read_users(user_service = Depends(get_user_service)):
return {"user_service": user_service}
# For testing, we override the dependencies
client = TestClient(app)
async def override_get_db():
return {"data": "Test database connection"}
app.dependency_overrides[get_db] = override_get_db
# The entire chain is affected
# get_user_service will now receive the test database
When testing, the get_db
dependency is replaced with override_get_db
, and this change propagates through the dependency chain to get_user_service
.
Practical Example: User Authentication System
Let's build a more complete example with sub-dependencies for a user authentication system:
from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from datetime import datetime, timedelta
from typing import Optional
import jwt
from pydantic import BaseModel
# Setup app
app = FastAPI()
# Secret key for JWT encoding/decoding
SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
# Mock database
fake_users_db = {
"johndoe": {
"username": "johndoe",
"full_name": "John Doe",
"email": "[email protected]",
"hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW", # "secret"
"disabled": False,
"roles": ["user"]
},
"alice": {
"username": "alice",
"full_name": "Alice Smith",
"email": "[email protected]",
"hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW", # "secret"
"disabled": False,
"roles": ["user", "admin"]
}
}
# Models
class Token(BaseModel):
access_token: str
token_type: str
class TokenData(BaseModel):
username: Optional[str] = None
class User(BaseModel):
username: str
email: Optional[str] = None
full_name: Optional[str] = None
disabled: Optional[bool] = None
roles: list = []
# First level dependency: OAuth scheme
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
# First level dependency: Token decoder
def decode_token(token: str = Depends(oauth2_scheme)):
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
username: str = payload.get("sub")
if username is None:
raise credentials_exception
token_data = TokenData(username=username)
except jwt.PyJWTError:
raise credentials_exception
return token_data
# Second level dependency: Get user from token
def get_user(token_data: TokenData = Depends(decode_token)):
user = fake_users_db.get(token_data.username)
if user is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="User not found"
)
return User(**user)
# Third level dependency: Verify user is active
def get_active_user(current_user: User = Depends(get_user)):
if current_user.disabled:
raise HTTPException(status_code=400, detail="Inactive user")
return current_user
# Fourth level dependency: Verify user is admin
def get_admin_user(current_user: User = Depends(get_active_user)):
if "admin" not in current_user.roles:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="User does not have admin privileges"
)
return current_user
# Create access token function (helper for login endpoint)
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
to_encode = data.copy()
if expires_delta:
expire = datetime.utcnow() + expires_delta
else:
expire = datetime.utcnow() + timedelta(minutes=15)
to_encode.update({"exp": expire})
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
return encoded_jwt
# Routes
@app.post("/token", response_model=Token)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
user = fake_users_db.get(form_data.username)
if not user or form_data.password != "secret": # In real app, verify hashed password
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect username or password",
headers={"WWW-Authenticate": "Bearer"},
)
access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
access_token = create_access_token(
data={"sub": user["username"]}, expires_delta=access_token_expires
)
return {"access_token": access_token, "token_type": "bearer"}
@app.get("/users/me/", response_model=User)
async def read_users_me(current_user: User = Depends(get_active_user)):
return current_user
@app.get("/admin/")
async def read_admin_data(admin_user: User = Depends(get_admin_user)):
return {
"admin_access": True,
"message": f"Hello admin {admin_user.full_name}",
"admin_roles": admin_user.roles
}
This example demonstrates how to build a complete dependency chain:
oauth2_scheme
extracts the token from the authorization headerdecode_token
validates and decodes the JWT tokenget_user
fetches the user based on the token dataget_active_user
checks if the user is activeget_admin_user
verifies admin privileges
Best Practices for Sub-dependencies
- Keep dependencies focused: Each dependency should have a single responsibility.
- Organize related dependencies: Group related dependencies in the same module.
- Use descriptive names: Name your functions to clearly indicate what they do.
- Consider using dependency classes: For complex logic, use classes to encapsulate behavior.
- Design for testability: Make your dependencies easy to mock or override for testing.
Common Pitfalls
- Circular dependencies: Avoid creating circular references between dependencies.
- Performance concerns: Be mindful that deep dependency chains might impact performance.
- Error handling: Ensure that errors are properly raised and handled throughout the dependency chain.
Summary
Sub-dependencies in FastAPI provide a powerful way to build modular, reusable components in your API. By chaining dependencies together, you can:
- Build complex authentication and authorization systems
- Separate concerns like validation, processing, and business logic
- Create reusable components that can be mixed and matched
- Make your code more maintainable and testable
The ability to override dependencies further enhances testability, allowing you to replace parts of your dependency chain during testing without modifying your actual code.
Additional Resources
- FastAPI Official Documentation on Dependencies
- FastAPI Advanced Dependency Injection
- JWT Authentication with FastAPI
Exercises
- Create a dependency chain that validates, sanitizes, and processes a search query parameter.
- Build a role-based authorization system with at least three levels (unauthenticated, user, admin).
- Implement a database dependency that can be easily overridden for testing.
- Create a logging system using dependencies that tracks request information across your application.
Happy coding with FastAPI sub-dependencies!
If you spot any mistakes on this website, please let me know at [email protected]. I’d greatly appreciate your feedback! :)