Understanding Dependency Injection in FastAPI (Async)
Dependency Injection (DI) is a powerful feature in FastAPI that allows you to manage and share resources efficiently, particularly in asynchronous applications. This guide will show how to leverage DI to manage async dependencies like database sessions and external APIs.
What Is Dependency Injection?
Dependency Injection is a design pattern where dependencies (e.g., database sessions, external service clients) are provided to a function or class, rather than being created inside the function or class itself.
In FastAPI, async dependencies can be used to optimize performance for high-concurrency applications.
Why Use Dependency Injection?
Here are the benefits of using Dependency Injection in FastAPI:
- Reusability: Write dependencies once and reuse them across multiple endpoints.
- Modularity: Keep your code clean and separated.
- Testability: Inject mock dependencies during testing.
- Performance: Async dependencies ensure optimal performance under high loads.
Setting Up Your Environment
Prerequisites
Ensure you have the following installed:
- Python 3.7 or higher
- FastAPI:
- Uvicorn (for running the server):
- SQLAlchemy (for database examples):
Basic Async Dependency
Here’s a simple async dependency example that returns a message:
from fastapi import FastAPI, Depends
app = FastAPI()
async def common_async_dependency():
return {"message": "This is an async dependency"}
@app.get("/")
async def read_root(dep: dict = Depends(common_async_dependency)):
return {"info": dep}
Async Dependency for Shared Resources
A common use case is managing an async database session.
Example: Async Database Dependency
from fastapi import FastAPI, Depends
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker
DATABASE_URL = "sqlite+aiosqlite:///./test.db"
engine = create_async_engine(DATABASE_URL, echo=True)
async_sessionmaker = sessionmaker(bind=engine, class_=AsyncSession, expire_on_commit=False)
async def get_async_db():
async with async_sessionmaker() as session:
yield session
app = FastAPI()
@app.get("/items/")
async def read_items(db: AsyncSession = Depends(get_async_db)):
return {"message": "Async database session provided!"}
Async Class-Based Dependencies
For managing state or configurations, use an async class-based dependency.
Example: Async Configuration Dependency
from fastapi import FastAPI, Depends
class Config:
def __init__(self, api_key: str):
self.api_key = api_key
async def get_async_config() -> Config:
return Config(api_key="SECRET_API_KEY")
app = FastAPI()
@app.get("/config/")
async def get_config_info(config: Config = Depends(get_async_config)):
return {"api_key": config.api_key}
Combining Multiple Async Dependencies
You can combine multiple async dependencies for a single endpoint.
Example: Combining Dependencies
from fastapi import FastAPI, Depends
async def dependency_one():
return {"dependency_one": "value1"}
async def dependency_two():
return {"dependency_two": "value2"}
app = FastAPI()
@app.get("/combined/")
async def combined(
dep_one: dict = Depends(dependency_one),
dep_two: dict = Depends(dependency_two)
):
return {"combined": {**dep_one, **dep_two}}
Wrapping Up
You’ve now learned how to use async dependency injection in FastAPI to manage shared resources, external services, and configurations. Dependency Injection makes your FastAPI application modular, reusable, and optimized for high-performance async workflows.
For more advanced examples, like integrating async dependencies with authentication or external APIs, check out Craft Your Startup.