Skip to main content

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:

  1. Check if the dependency has already been called with the same parameters in the current request
  2. If it has, return the previously computed value
  3. 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:

python
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:

python
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/, both c1 and c2 will have the same value because the dependency is cached by default.
  • When you access /uncached-counter/, c1 and c2 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:

python
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:

python
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:

python
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:

python
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:

python
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:

  1. The dependency performs expensive computations
  2. The dependency makes database queries
  3. The dependency makes external API calls
  4. The dependency is used multiple times in a single request

When to Disable Caching

You might want to disable caching when:

  1. You need fresh data for each call
  2. The dependency has side effects that should happen each time
  3. 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

Exercises

  1. Create a FastAPI application that demonstrates caching with a simulated expensive computation (e.g., a function with a sleep delay).
  2. Implement a database dependency that uses caching to avoid multiple connections within one request.
  3. Create a dependency system that fetches user preferences from an external service and caches the results.
  4. Implement a caching system that varies based on a parameter, showing when dependencies are and aren't cached.
  5. 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! :)