FastAPI Dependency Caching
Introduction
When building web applications with FastAPI, you often need to reuse the same functionality across multiple endpoints. Dependency injection provides a clean way to share code between route handlers, but sometimes these dependencies perform expensive operations that shouldn't be repeated unnecessarily within the same request. This is where dependency caching comes in.
Dependency caching allows you to compute a value once and reuse it multiple times within the same request scope, improving performance and reducing redundant operations. In this guide, we'll explore how FastAPI handles dependency caching, when to use it, and how to implement it effectively.
Understanding Dependency Caching in FastAPI
FastAPI automatically caches dependencies with the same parameters within the same request. This means that if multiple route operations or other dependencies depend on the same dependency function, that function will only be executed once per request.
How Caching Works
By default, when you declare a dependency using the Depends()
function, FastAPI will:
- Check if the dependency has already been called with the same parameters in the current request
- If it has, return the previously computed value
- If not, call the dependency function and cache its result for future use within the same request
Basic Example of Dependency Caching
Let's look at a simple example to understand how dependency caching works:
from fastapi import FastAPI, Depends
app = FastAPI()
async def get_query_params(q: str = None):
print(f"Processing query parameter: {q}")
return {"q": q}
@app.get("/items/")
async def read_items(query_params: dict = Depends(get_query_params)):
return {"query_params": query_params, "message": "This is the items endpoint"}
@app.get("/users/")
async def read_users(query_params: dict = Depends(get_query_params)):
return {"query_params": query_params, "message": "This is the users endpoint"}
When you make a request to either /items/?q=test
or /users/?q=test
, the get_query_params
function will be called only once per request, even though both route handlers depend on it. You'll see the "Processing query parameter" message printed only once in your console.
Controlling Caching Behavior
Sometimes you might want to disable caching for certain dependencies. FastAPI allows you to control this behavior using the use_cache
parameter in the Depends()
function.
Disabling Caching
You can disable caching for a dependency by setting use_cache=False
:
from fastapi import FastAPI, Depends
app = FastAPI()
counter = 0
def get_counter():
global counter
counter += 1
print(f"Counter is now: {counter}")
return counter
@app.get("/cached-counter/")
async def read_cached_counter(c1: int = Depends(get_counter), c2: int = Depends(get_counter)):
return {"counter1": c1, "counter2": c2} # c1 and c2 will be the same value
@app.get("/uncached-counter/")
async def read_uncached_counter(
c1: int = Depends(get_counter),
c2: int = Depends(get_counter, use_cache=False)
):
return {"counter1": c1, "counter2": c2} # c1 and c2 will be different values
In this example:
- When you access
/cached-counter/
, bothc1
andc2
will have the same value because the dependency is cached by default. - When you access
/uncached-counter/
,c1
andc2
will have different values because we explicitly disabled caching for the second dependency.
Practical Use Cases for Dependency Caching
Let's explore some real-world scenarios where dependency caching provides significant benefits:
1. Database Connection Management
Caching a database connection within a request scope can prevent multiple connections from being established for the same request:
from fastapi import FastAPI, Depends
import databases
import os
app = FastAPI()
DATABASE_URL = os.getenv("DATABASE_URL", "sqlite:///./test.db")
async def get_database():
print("Creating database connection...")
database = databases.Database(DATABASE_URL)
await database.connect()
try:
yield database
finally:
print("Closing database connection...")
await database.disconnect()
@app.get("/users/{user_id}")
async def read_user(user_id: int, db: databases.Database = Depends(get_database)):
query = "SELECT * FROM users WHERE id = :id"
user = await db.fetch_one(query=query, values={"id": user_id})
# Another operation using the same connection
query = "SELECT COUNT(*) as post_count FROM posts WHERE user_id = :id"
post_count = await db.fetch_one(query=query, values={"id": user_id})
return {"user": dict(user), "post_count": dict(post_count)}
In this example, both database queries use the same cached database connection, improving performance and resource utilization.
2. Authentication and User Data Retrieval
When implementing authentication, you often need to retrieve user information once and use it in multiple places:
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
import jwt
from datetime import datetime
app = FastAPI()
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
fake_users_db = {
"johndoe": {
"username": "johndoe",
"full_name": "John Doe",
"email": "[email protected]",
"hashed_password": "fakehashedsecret",
"disabled": False,
}
}
class User:
def __init__(self, username: str, email: str, full_name: str, disabled: bool):
self.username = username
self.email = email
self.full_name = full_name
self.disabled = disabled
async def get_current_user(token: str = Depends(oauth2_scheme)):
print("Decoding JWT token...")
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
username: str = payload.get("sub")
if username is None:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
except jwt.PyJWTError:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
user_data = fake_users_db.get(username)
if user_data is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)
return User(
username=user_data["username"],
email=user_data["email"],
full_name=user_data["full_name"],
disabled=user_data["disabled"]
)
async def get_current_active_user(current_user: User = Depends(get_current_user)):
if current_user.disabled:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST)
return current_user
@app.get("/users/me")
async def read_users_me(current_user: User = Depends(get_current_active_user)):
return current_user
@app.get("/users/me/items")
async def read_own_items(current_user: User = Depends(get_current_active_user)):
return [{"item_id": 1, "owner": current_user.username}]
In this example, the JWT token decoding and user retrieval only happen once per request, even when both route handlers depend on the get_current_active_user
function, which in turn depends on get_current_user
.
3. External API Calls
When your application needs to make external API calls, caching these calls within a request can prevent redundant network requests:
from fastapi import FastAPI, Depends
import httpx
import time
app = FastAPI()
async def get_weather_data(city: str = "London"):
print(f"Fetching weather data for {city}...")
async with httpx.AsyncClient() as client:
# Simulate an API call with a delay
response = await client.get(f"https://api.example.com/weather/{city}")
# We're simulating a response here
return {
"city": city,
"temperature": 22,
"conditions": "Sunny",
"timestamp": time.time()
}
@app.get("/weather/{city}")
async def read_weather(
city: str,
weather_data: dict = Depends(get_weather_data)
):
# The external API call is made only once
return {
"current_weather": weather_data,
"forecast_summary": f"Weather in {city} looks great!"
}
@app.get("/travel/{city}")
async def travel_recommendations(
city: str,
weather_data: dict = Depends(get_weather_data)
):
# Reuses the cached weather data without making another API call
if weather_data["temperature"] > 20:
recommendation = "Great time to visit!"
else:
recommendation = "Better bring a jacket!"
return {
"city": city,
"recommendation": recommendation,
"current_weather": weather_data
}
In this example, when a user requests both endpoints for the same city in a single request, the external API call is made only once, improving response times and reducing load on the external API.
Advanced Caching Patterns
Caching with Parameters
FastAPI caches dependencies based on their parameters. If you call the same dependency with different parameters, it will execute the function multiple times:
from fastapi import FastAPI, Depends
app = FastAPI()
async def get_item_details(item_id: int):
print(f"Fetching details for item {item_id}")
return {"item_id": item_id, "name": f"Item {item_id}"}
@app.get("/items/{item_id}")
async def read_item(
item_id: int,
item: dict = Depends(get_item_details),
related_item: dict = Depends(get_item_details)
):
# This will call get_item_details once with item_id
# And once with item_id (from the path) for related_item
# But because both calls use the same parameter, it's cached
return {
"item": item,
"related": related_item
}
@app.get("/compare/")
async def compare_items(
item1_id: int = 1,
item2_id: int = 2,
item1: dict = Depends(lambda: get_item_details(item1_id)),
item2: dict = Depends(lambda: get_item_details(item2_id))
):
# This will call get_item_details twice with different parameters
return {
"item1": item1,
"item2": item2,
"same": item1 == item2
}
Using Sub-Dependencies
Dependencies can depend on other dependencies, and FastAPI will maintain proper caching for the entire dependency tree:
from fastapi import FastAPI, Depends, Header
from typing import Optional
app = FastAPI()
async def verify_token(x_token: Optional[str] = Header(None)):
print("Verifying token...")
if x_token != "fake-super-secret-token":
raise HTTPException(status_code=400, detail="X-Token header invalid")
return x_token
async def verify_key(x_key: Optional[str] = Header(None)):
print("Verifying key...")
if x_key != "fake-super-secret-key":
raise HTTPException(status_code=400, detail="X-Key header invalid")
return x_key
async def get_auth_dependencies(
token: str = Depends(verify_token),
key: str = Depends(verify_key)
):
print("Getting auth dependencies...")
return {"token": token, "key": key}
@app.get("/items/")
async def read_items(auth: dict = Depends(get_auth_dependencies)):
return {
"auth": auth,
"items": [
{"item_id": "Foo"},
{"item_id": "Bar"}
]
}
@app.get("/users/")
async def read_users(auth: dict = Depends(get_auth_dependencies)):
return {
"auth": auth,
"users": [
{"user_id": 1},
{"user_id": 2}
]
}
In this example, when a request is made to either /items/
or /users/
, the verify_token
and verify_key
functions are only called once per request, even though both endpoints use the get_auth_dependencies
function.
Performance Considerations
When to Use Caching
Dependency caching is particularly useful when:
- The dependency performs expensive computations
- The dependency makes database queries
- The dependency makes external API calls
- The dependency is used multiple times in a single request
When to Disable Caching
You might want to disable caching when:
- You need fresh data for each call
- The dependency has side effects that should happen each time
- The dependency function updates a value or state
Summary
FastAPI's dependency caching mechanism provides an elegant way to optimize your application's performance by avoiding redundant computations within a request. Key points to remember:
- Dependencies are cached by default within the same request
- Caching is based on the dependency function and its parameters
- You can disable caching with
use_cache=False
when needed - Caching works throughout the dependency tree, ensuring optimal performance
- Use caching strategically for expensive operations like database queries and API calls
By understanding how dependency caching works in FastAPI, you can design more efficient APIs that respond quickly while minimizing resource usage.
Additional Resources
- FastAPI Official Documentation on Dependencies
- Advanced Dependency Injection Patterns in FastAPI
- Python's Functional Programming Tools
Exercises
- Create a FastAPI application that demonstrates caching with a simulated expensive computation (e.g., a function with a sleep delay).
- Implement a database dependency that uses caching to avoid multiple connections within one request.
- Create a dependency system that fetches user preferences from an external service and caches the results.
- Implement a caching system that varies based on a parameter, showing when dependencies are and aren't cached.
- Design a FastAPI application that combines multiple cached dependencies in a complex hierarchy.
If you spot any mistakes on this website, please let me know at [email protected]. I’d greatly appreciate your feedback! :)