FastAPI is a modern, fast web framework for building APIs with Python. One of its powerful features is dependency injection, which allows for modular and flexible application design. Let’s dive into how to implement dependency injection with complex dependencies in FastAPI.

Dependency injection is a technique where objects are passed into a class or function instead of being created inside. This promotes loose coupling and makes our code more maintainable and testable. In FastAPI, we can use dependencies to inject objects into our route handlers.

To get started with dependency injection in FastAPI, we first need to define our dependencies. These can be simple functions or more complex classes. Let’s look at a basic example:

from fastapi import Depends

def get_db():
    db = Database()
    try:
        yield db
    finally:
        db.close()

@app.get("/items")
async def read_items(db: Database = Depends(get_db)):
    return db.get_items()

In this example, we define a get_db function that yields a database connection. We then use the Depends function to inject this dependency into our route handler.

But what about more complex dependencies? Let’s say we have a user authentication system and we want to inject the current user into our routes. We can create a dependency that checks for a valid token and returns the user:

from fastapi import Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

def get_current_user(token: str = Depends(oauth2_scheme)):
    user = decode_token(token)
    if not user:
        raise HTTPException(status_code=401, detail="Invalid token")
    return user

@app.get("/users/me")
async def read_users_me(current_user: User = Depends(get_current_user)):
    return current_user

Here, we’re using FastAPI’s built-in OAuth2 support to handle token-based authentication. The get_current_user function is a dependency that checks the token and returns the user.

Now, let’s take it a step further and implement a more complex dependency system. Imagine we’re building an e-commerce platform and we need to handle product inventory, user carts, and order processing. We can create a series of interdependent classes to manage this:

class InventoryManager:
    def __init__(self, db: Database):
        self.db = db

    def check_stock(self, product_id: int, quantity: int):
        stock = self.db.get_stock(product_id)
        return stock >= quantity

class CartManager:
    def __init__(self, inventory: InventoryManager):
        self.inventory = inventory

    def add_to_cart(self, user_id: int, product_id: int, quantity: int):
        if self.inventory.check_stock(product_id, quantity):
            # Add to cart logic here
            return True
        return False

class OrderProcessor:
    def __init__(self, cart: CartManager, db: Database):
        self.cart = cart
        self.db = db

    def process_order(self, user_id: int):
        cart_items = self.db.get_cart_items(user_id)
        for item in cart_items:
            if not self.cart.add_to_cart(user_id, item.product_id, item.quantity):
                return False
        # Process payment and create order
        return True

def get_order_processor(db: Database = Depends(get_db)):
    inventory = InventoryManager(db)
    cart = CartManager(inventory)
    return OrderProcessor(cart, db)

@app.post("/order")
async def create_order(order_processor: OrderProcessor = Depends(get_order_processor), current_user: User = Depends(get_current_user)):
    success = order_processor.process_order(current_user.id)
    if success:
        return {"message": "Order placed successfully"}
    else:
        raise HTTPException(status_code=400, detail="Failed to place order")

In this example, we’ve created a complex dependency chain. The OrderProcessor depends on CartManager, which depends on InventoryManager, which depends on the database connection. We’ve encapsulated all this logic in the get_order_processor function, which FastAPI will use to resolve the dependencies.

This approach allows us to keep our route handlers clean and focused on their specific tasks, while the complex business logic is handled by our dependency classes.

One of the great things about FastAPI’s dependency injection system is that it’s not limited to route handlers. We can use dependencies in middleware, background tasks, and even other dependencies. This allows for incredible flexibility in how we structure our applications.

For instance, we could create a dependency that logs all requests:

import time
from fastapi import Request

async def log_request(request: Request):
    start_time = time.time()
    await request.json()
    process_time = time.time() - start_time
    print(f"Request to {request.url} took {process_time:.2f} seconds")

@app.get("/items", dependencies=[Depends(log_request)])
async def read_items():
    return {"items": ["foo", "bar"]}

This dependency will log the processing time for each request to the /items endpoint.

We can also use dependencies to implement role-based access control:

from enum import Enum

class Role(str, Enum):
    USER = "user"
    ADMIN = "admin"

def check_role(required_role: Role):
    def inner(current_user: User = Depends(get_current_user)):
        if current_user.role != required_role:
            raise HTTPException(status_code=403, detail="Not enough permissions")
        return current_user
    return inner

@app.get("/admin", dependencies=[Depends(check_role(Role.ADMIN))])
async def admin_route():
    return {"message": "Welcome, admin!"}

This check_role dependency factory allows us to easily restrict access to certain routes based on the user’s role.

One of the most powerful aspects of FastAPI’s dependency injection system is its ability to handle asynchronous dependencies. This is particularly useful when working with asynchronous databases or external APIs. Let’s look at an example:

import aiohttp
from fastapi import Depends

async def get_external_data():
    async with aiohttp.ClientSession() as session:
        async with session.get('https://api.example.com/data') as response:
            return await response.json()

@app.get("/external-data")
async def read_external_data(data: dict = Depends(get_external_data)):
    return data

In this example, we’re using an asynchronous HTTP client to fetch data from an external API. FastAPI will automatically await this dependency, ensuring that our application remains non-blocking and efficient.

Another advanced use case for dependency injection is implementing a caching layer. We can create a dependency that checks a cache before hitting the database:

import redis
from fastapi import Depends

redis_client = redis.Redis()

class CachedDatabase:
    def __init__(self, db: Database):
        self.db = db

    async def get_item(self, item_id: int):
        # Check cache first
        cached_item = redis_client.get(f"item:{item_id}")
        if cached_item:
            return cached_item

        # If not in cache, get from DB and cache it
        item = await self.db.get_item(item_id)
        redis_client.set(f"item:{item_id}", item)
        return item

def get_cached_db(db: Database = Depends(get_db)):
    return CachedDatabase(db)

@app.get("/items/{item_id}")
async def read_item(item_id: int, db: CachedDatabase = Depends(get_cached_db)):
    return await db.get_item(item_id)

This setup allows us to transparently add caching to our database queries without cluttering our route handlers.

Dependency injection in FastAPI isn’t just about simplifying our code structure – it’s also about improving performance. FastAPI is smart about how it handles dependencies. If multiple route handlers in a request use the same dependency, FastAPI will only execute that dependency once and reuse the result. This can significantly reduce the overhead in our applications.

Moreover, FastAPI’s dependency system integrates seamlessly with its automatic API documentation. Any parameters required by our dependencies will be automatically included in the OpenAPI schema, making it easy for clients to understand what data they need to provide.

As our applications grow in complexity, we might find ourselves with a large number of dependencies. FastAPI allows us to organize these into dependency containers:

from fastapi import Depends

class ServiceContainer:
    def __init__(self):
        self.db = Database()
        self.cache = Redis()
        self.external_api = ExternalAPI()

container = ServiceContainer()

def get_db(services: ServiceContainer = Depends(lambda: container)):
    return services.db

def get_cache(services: ServiceContainer = Depends(lambda: container)):
    return services.cache

@app.get("/items")
async def read_items(db: Database = Depends(get_db), cache: Redis = Depends(get_cache)):
    # Use db and cache here
    pass

This approach allows us to manage all our services in one place, making it easier to handle things like database connections or API clients.

In conclusion, FastAPI’s dependency injection system is a powerful tool for building modular, efficient, and maintainable applications. By leveraging dependencies, we can separate concerns, reuse code, and create clean, intuitive APIs. Whether we’re working with simple database connections or complex business logic, dependency injection helps us write better, more testable code. As we’ve seen, it integrates seamlessly with FastAPI’s other features like async support and API documentation, making it an essential part of any FastAPI developer’s toolkit. So next time you’re building a FastAPI application, think about how you can use dependency injection to improve your code structure and make your life easier!