FastAPI Dependency Management

When building APIs with FastAPI, one of its most powerful features is the dependency injection system. This elegant mechanism helps you organize your code, make it more maintainable, and avoid repetition. In this tutorial, we'll explore how to effectively use FastAPI's dependency management to write better code.

What are Dependencies in FastAPI?

Dependencies in FastAPI are functions (or callable objects) that can be used to:

1. Share code logic between different path operation functions
2. Implement security and authentication flows
3. Handle database connections
4. Validate additional inputs or parameters
5. Return specific data or objects that will be used by path operations

When you declare a dependency for a path operation, FastAPI will automatically execute the dependency function before your path operation function and pass its results to your function.

Basic Dependency Example

Let's start with a simple example:

python

from fastapi import Depends, FastAPI

app = FastAPI()

async def get_query_parameter(q: str = None):
    return {"q": q}

@app.get("/items/")
async def read_items(query_params: dict = Depends(get_query_parameter)):
    return query_params

Here's what happens when you call this endpoint:

1. When a request comes to /items/, FastAPI executes get_query_parameter() first
2. If the request has a query parameter q, it gets passed to the dependency
3. The dependency returns a dictionary with the query parameter
4. This dictionary is then passed to the read_items() function as query_params
5. The path operation function returns this dictionary

If you make a request to /items/?q=test, you'll get this response:

json

{
  "q": "test"
}

Creating Reusable Dependencies

One of the main benefits of dependencies is reusability. Let's create a more practical example:

python

from fastapi import Depends, FastAPI, HTTPException
from typing import Annotated

app = FastAPI()

# A simple database simulation
fake_users_db = {
    "john": {"username": "john", "email": "[email protected]", "is_active": True},
    "alice": {"username": "alice", "email": "[email protected]", "is_active": True},
    "bob": {"username": "bob", "email": "[email protected]", "is_active": False},
}

def get_user(username: str):
    """Dependency to fetch user data from database"""
    if username not in fake_users_db:
        raise HTTPException(status_code=404, detail="User not found")
    return fake_users_db[username]

def get_active_user(user: dict = Depends(get_user)):
    """Dependency that uses another dependency"""
    if not user["is_active"]:
        raise HTTPException(status_code=400, detail="Inactive user")
    return user

# Type annotations can make dependencies clearer
UserDep = Annotated[dict, Depends(get_user)]
ActiveUserDep = Annotated[dict, Depends(get_active_user)]

@app.get("/users/{username}")
async def read_user(user: UserDep):
    return user

@app.get("/active-users/{username}")
async def read_active_user(user: ActiveUserDep):
    return user

In this example:

1. We define a get_user dependency that retrieves a user from our fake database
2. We then create a get_active_user dependency that depends on get_user and adds additional logic
3. We use type annotations to make our code cleaner
4. We use these dependencies in different path operations

If you call /users/john, you'll get John's data. If you call /users/bob, you'll also get Bob's data. But if you call /active-users/bob, you'll get a 400 error because Bob is not active.

Class-based Dependencies

For more complex dependencies, you can use classes that implement the __call__ method:

python

from fastapi import Depends, FastAPI
from typing import Annotated

app = FastAPI()

class QueryChecker:
    def __init__(self, min_length: int = 3):
        self.min_length = min_length
        
    def __call__(self, q: str = None):
        if q is None:
            return None
        if len(q) < self.min_length:
            return f"Query is too short (min: {self.min_length})"
        return q

# Create instances with different configurations
default_checker = QueryChecker()
strict_checker = QueryChecker(min_length=5)

@app.get("/default-check/")
async def read_with_default_check(result: Annotated[str, Depends(default_checker)]):
    return {"result": result}

@app.get("/strict-check/")
async def read_with_strict_check(result: Annotated[str, Depends(strict_checker)]):
    return {"result": result}

Class-based dependencies are especially useful when you need to:

• Configure the dependency with parameters
• Reuse the dependency with different configurations
• Organize related dependencies together
• Have a more complex state or behavior

Global Dependencies

Sometimes you need to execute certain code for all requests. FastAPI allows you to add dependencies to the application itself:

python

from fastapi import Depends, FastAPI, Header, HTTPException
from typing import Annotated

async def verify_api_key(x_api_key: Annotated[str | None, Header()] = None):
    if x_api_key != "valid_api_key":
        raise HTTPException(status_code=403, detail="Invalid API Key")
    return x_api_key

app = FastAPI(dependencies=[Depends(verify_api_key)])

@app.get("/items/")
async def read_items():
    return {"message": "You have access to the items"}

@app.get("/users/")
async def read_users():
    return {"message": "You have access to the users"}

Now, both endpoints require a valid API key in the header (X-API-Key: valid_api_key).

Dependency Design Patterns

1. Sub-dependencies (dependency trees)

You can create chains of dependencies where one dependency depends on another:

python

from fastapi import Depends, FastAPI
from typing import Annotated

app = FastAPI()

async def common_parameters(q: str = None, skip: int = 0, limit: int = 100):
    return {"q": q, "skip": skip, "limit": limit}

async def pagination(commons: Annotated[dict, Depends(common_parameters)]):
    return {"skip": commons["skip"], "limit": commons["limit"]}

async def filtering(commons: Annotated[dict, Depends(common_parameters)]):
    return {"q": commons["q"]}

@app.get("/items/")
async def read_items(
    pagination_params: Annotated[dict, Depends(pagination)],
    filter_params: Annotated[dict, Depends(filtering)]
):
    return {
        "pagination": pagination_params,
        "filters": filter_params,
        "items": [{"item_id": i} for i in range(pagination_params["skip"], pagination_params["skip"] + pagination_params["limit"])]
    }

2. Yield Dependencies for Resource Management

FastAPI supports dependencies that use yield instead of return, which is perfect for resource management:

python

from fastapi import Depends, FastAPI
from typing import Annotated

app = FastAPI()

async def get_db_connection():
    print("Opening database connection")
    # In a real application, this would connect to a database
    db = {"connection": "object"}
    yield db
    print("Closing database connection")
    # In a real application, this would close the connection

@app.get("/items/")
async def read_items(db: Annotated[dict, Depends(get_db_connection)]):
    # Use db connection here
    print("Using database connection")
    return {"message": "Database connection was used"}

When you call this endpoint, the logs will show:

Opening database connection
Using database connection
Closing database connection

This is extremely useful for:

• Database connections
• File handling
• Network connections
• Any resource that needs proper cleanup

Solving Common Problems with Dependencies

Problem 1: Authentication and Authorization

python

from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from typing import Annotated

app = FastAPI()

# This is a simplified auth system - in a real app, you'd validate tokens properly
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

fake_users_db = {
    "john": {
        "username": "john",
        "email": "[email protected]",
        "is_admin": False
    },
    "alice": {
        "username": "alice",
        "email": "[email protected]",
        "is_admin": True
    }
}

# Token simulation - in a real app you'd decode and verify JWT tokens
fake_tokens = {
    "john_token": "john",
    "alice_token": "alice"
}

async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]):
    if token not in fake_tokens:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid authentication credentials",
            headers={"WWW-Authenticate": "Bearer"},
        )
    
    username = fake_tokens[token]
    user = fake_users_db.get(username)
    
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="User not found",
            headers={"WWW-Authenticate": "Bearer"},
        )
    
    return user

async def get_admin_user(current_user: Annotated[dict, Depends(get_current_user)]):
    if not current_user["is_admin"]:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN, 
            detail="Not enough permissions"
        )
    return current_user

@app.get("/users/me")
async def read_users_me(current_user: Annotated[dict, Depends(get_current_user)]):
    return current_user

@app.get("/admin-only")
async def read_admin_data(admin_user: Annotated[dict, Depends(get_admin_user)]):
    return {"message": "You are an admin", "user": admin_user}

Problem 2: Database Session Management

python

from fastapi import Depends, FastAPI, HTTPException
from typing import Annotated, List
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session

app = FastAPI()

# In a real app, you would use environment variables for the connection string
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

# Dependency for database sessions
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# Models would be defined here in a real app

@app.get("/users/")
async def read_users(db: Annotated[Session, Depends(get_db)]):
    # In a real app, you would query the database
    # users = db.query(User).all()
    # For this example, we'll just return dummy data
    return [{"id": 1, "name": "John"}, {"id": 2, "name": "Alice"}]

@app.get("/users/{user_id}")
async def read_user(user_id: int, db: Annotated[Session, Depends(get_db)]):
    # In a real app, you would query the database
    # user = db.query(User).filter(User.id == user_id).first()
    # if user is None:
    #     raise HTTPException(status_code=404, detail="User not found")
    
    # For this example, we'll just return dummy data
    if user_id not in [1, 2]:
        raise HTTPException(status_code=404, detail="User not found")
    
    user_data = {"id": user_id, "name": "John" if user_id == 1 else "Alice"}
    return user_data

Using Dependency Overrides for Testing

FastAPI allows you to override dependencies during testing, which is incredibly useful:

python

from fastapi import Depends, FastAPI
from fastapi.testclient import TestClient
from typing import Annotated

app = FastAPI()

async def get_db():
    # This would normally connect to the real database
    return {"name": "production_db", "data": "real data"}

@app.get("/items/")
async def read_items(db: Annotated[dict, Depends(get_db)]):
    return {"database": db["name"], "data": db["data"]}

# For testing, we can override the dependency
client = TestClient(app)

async def override_get_db():
    # Return a test database instead
    return {"name": "test_db", "data": "test data"}

# Test function
def test_read_items():
    # Override the dependency
    app.dependency_overrides[get_db] = override_get_db
    
    # Make the request
    response = client.get("/items/")
    
    # Assert the response
    assert response.status_code == 200
    assert response.json() == {"database": "test_db", "data": "test data"}
    
    # Clean up the override
    app.dependency_overrides = {}

This feature makes unit testing much easier since you can replace real dependencies (like databases, external APIs, etc.) with test doubles.

Summary

FastAPI's dependency injection system is a powerful tool for writing clean, maintainable, and testable code. We've explored:

• Basic dependency usage
• Creating dependency chains
• Class-based dependencies
• Global dependencies
• Resource management with yield dependencies
• Common patterns for authentication, database management, and testing

These techniques will help you structure your FastAPI applications better, reduce code duplication, and make your code more modular and testable.

Additional Resources

1. Official FastAPI Dependencies Documentation
2. FastAPI Security Guide
3. SQLAlchemy with FastAPI

Exercises

1. Create a FastAPI application with a dependency that validates that a header X-Request-ID is present in all requests.

2. Implement a rate-limiting dependency that allows only 5 requests per minute from the same IP address.

3. Create a dependency tree where:

   • A base dependency gets query parameters
   • A second dependency validates them
   • A third dependency transforms the parameters
   • Finally, use all these in a path operation

4. Implement a context management dependency using yield that tracks the time it takes to process each request and logs it.

5. Create a testing setup where you override database dependencies with mock implementations for unit testing.

By mastering dependency management in FastAPI, you'll be able to build more robust and maintainable APIs. Happy coding!