In the world of FastAPI, Dependency Injection (DI) is a core tool for managing resource sharing and reuse in code. Especially when handling asynchronous tasks, DI helps us gracefully manage “resources” such as database connections, authentication information, and configuration parameters, avoiding code duplication and coupling. This article will break down the core concepts and practical techniques of asynchronous dependency injection step by step, suitable for developers new to FastAPI.
一、What is Dependency Injection? Why is it Important?¶
Imagine you write a route function that needs to query a database to get user information. In the traditional approach, you might directly write create_db_connection() inside the function to get the connection and then execute the query. However, this approach has issues:
- Connection objects are scattered across functions, and repeated creation and closing waste resources.
- If the database connection logic changes (e.g., switching to an asynchronous driver), all functions relying on it need modification.
Core Idea of Dependency Injection: Hand over the acquisition and management of “resources” (e.g., database connections) to an external system. Functions only need to declare “I need this resource” instead of acquiring it themselves. This makes the code clearer, more decoupled, and easier to test and maintain.
二、Basics of Dependency Injection in FastAPI¶
FastAPI uses the Depends() tool to declare dependencies. The syntax is simple: parameter = Depends(dependency_function). Dependencies can be regular functions (synchronous) or asynchronous functions (async def), and FastAPI automatically handles the invocation logic.
1. Synchronous Dependency Injection (Basic)¶
Let’s look at a simple synchronous example, such as obtaining a database connection:
from fastapi import FastAPI, Depends
# Assume this is a synchronous database connection function
def get_sync_db():
db = create_sync_db_connection() # Synchronous connection creation
return db
app = FastAPI()
@app.get("/users")
def read_users(db=Depends(get_sync_db)):
# Use db to query user data (assuming db is a synchronous object)
return db.query("SELECT * FROM users")
Here, get_sync_db is a synchronous function. Depends(get_sync_db) tells FastAPI: “First execute get_sync_db(), and pass the result to the db parameter.”
2. Asynchronous Dependency Injection (Key)¶
When the dependency requires asynchronous operations (e.g., async database queries, async HTTP requests), simply define the dependency function as async def, and FastAPI will automatically call it with await.
from fastapi import FastAPI, Depends
import asyncio
# Asynchronous database connection function
async def get_async_db():
async with aio_db.AsyncConnection() as db: # Assume aio_db is an async database driver
yield db # Use yield to implement a context manager (optional, can also return directly)
# Alternatively, return the connection object directly
async def get_async_db():
db = await aio_db.connect() # Asynchronously connect to the database
return db
app = FastAPI()
@app.get("/users")
async def read_users(db=Depends(get_async_db)):
# Use await to call async database queries
return await db.fetch("SELECT * FROM users")
Key points:
- The dependency function must be defined with async def.
- Depends(get_async_db) will automatically await the function call and return the final db object (after asynchronous operations are completed).
三、Advanced Techniques for Asynchronous Dependency Injection¶
1. Nested and Composed Dependencies¶
In complex scenarios, one dependency may rely on another. For example, querying a user requires first obtaining the user ID, which comes from authentication information.
from fastapi import Depends, FastAPI, HTTPException
from pydantic import BaseModel
# 1. Authentication dependency: Get current user ID
def get_current_user_id():
# Assume get user ID from a Token (use a more secure method in actual scenarios)
return "user_123" # Example return user ID
# 2. Database query dependency: Depends on user ID to fetch user info
async def get_user_by_id(user_id=Depends(get_current_user_id)):
db = Depends(get_async_db) # Depend on the above database connection
user = await db.query(f"SELECT * FROM users WHERE id={user_id}")
if not user:
raise HTTPException(status_code=404, detail="User not found")
return user
app = FastAPI()
@app.get("/user")
async def get_user(user=Depends(get_user_by_id)):
return user
Here, get_user_by_id relies on both get_current_user_id (synchronous) and get_async_db (asynchronous). FastAPI automatically resolves dependencies in sequence to ensure correct invocation logic.
2. Dependency Passing in Asynchronous Tasks¶
When reusing dependencies in asynchronous tasks (e.g., background tasks, message queue tasks), simply pass the dependency parameters directly. FastAPI supports injecting dependencies in task functions.
from fastapi import BackgroundTasks, Depends, FastAPI
# 1. Asynchronous database connection dependency
async def get_db():
db = await aio_db.connect()
return db
# 2. Background task function (asynchronous)
async def send_welcome_email_task(db=Depends(get_db), user_id: str):
await db.execute(f"INSERT INTO emails (user_id, content) VALUES ({user_id}, 'Welcome!')")
app = FastAPI()
@app.post("/register")
async def register(
user_id: str,
background_tasks: BackgroundTasks,
db=Depends(get_db)
):
# 1. Register the user first (assuming db is for synchronous operations)
await db.execute(f"INSERT INTO users (id) VALUES ({user_id})")
# 2. Add an asynchronous background task, passing db and user_id
background_tasks.add_task(
send_welcome_email_task, # Task function
user_id=user_id # Directly pass the parameters required by the dependency
)
return {"message": "User registered, welcome email queued"}
Here, send_welcome_email_task receives the db dependency and user_id parameter. FastAPI automatically injects db (resolved via Depends(get_db)) when calling the task.
3. Avoid Common Pitfalls¶
- Forgetting to await asynchronous dependencies: If the dependency is
async defbut you return a coroutine object (not the result), subsequent calls will error. Always let FastAPI handleawaitviaDepends. - Circular dependencies: If A depends on B and B depends on A, FastAPI throws a
RuntimeError. Refactor code, split dependencies, or use caching. - Return type of dependencies: Ensure the dependency returns the expected object type (e.g., database connection, user info) to avoid type mismatches.
四、Summary¶
FastAPI’s asynchronous dependency injection, via the Depends() tool, allows graceful management of asynchronous resources (e.g., database connections, authentication). Key points:
1. Dependency Functions: Use regular functions for synchronous operations, async def for asynchronous ones. FastAPI automatically handles await.
2. Nested Dependencies: Declare multi-layer dependencies with Depends(), resolved in sequence by FastAPI.
3. Task Passing: Directly pass dependency parameters in asynchronous tasks to ensure resource reuse.
Mastering these techniques helps you build decoupled, scalable FastAPI applications efficiently, especially when handling asynchronous tasks, avoiding redundant code and resource waste.
Exercise: Try adding two dependencies (e.g., get_db and get_config) to an asynchronous route function and use them to complete a simple database query.