Fix: FastAPI Dependency Injection Errors — Dependencies Not Working
Quick Answer
How to fix FastAPI dependency injection errors — async dependencies, database sessions, sub-dependencies, dependency overrides in tests, and common DI mistakes.
The Error
FastAPI raises an error when resolving a dependency:
fastapi.exceptions.FastAPIError: Invalid args for response field!
Hint: check that typing.Optional[Session] is a valid Pydantic field type.Or a database session dependency doesn’t close properly:
async def get_db():
db = SessionLocal()
yield db
# db.close() never called if an exception occurs before yieldOr a dependency parameter type causes a validation error:
422 Unprocessable Entity
{
"detail": [{"loc": ["query", "current_user"], "msg": "field required", "type": "value_error.missing"}]
}
# FastAPI treated the Depends() parameter as a query parameterOr circular dependencies cause a startup error:
RuntimeError: Dependency 'get_current_user' is a coroutine function but it's not wrapped with async def in the function signature.Why This Happens
FastAPI’s dependency injection works by inspecting function signatures. Parameters typed with Depends() are resolved from other functions, not from request data. The most common mistakes:
- Missing
Depends()wrapper — a parameter typed with a dependency class/function but withoutDepends()is treated as a query parameter, causing a 422 error. - Sync dependency in async context (or vice versa) — FastAPI handles both, but mixing them incorrectly causes runtime errors or blocking behavior.
- Generator dependency not using
yield— dependencies that need cleanup (database sessions, file handles) must useyieldto ensure cleanup code runs after the request. yieldin an async generator withoutasync def— async generator dependencies must useasync def+yield, notdef+yield.- Class-based dependency missing
__call__— a class used as a dependency must be callable (implement__call__). - Response model including dependency types — SQLAlchemy
Sessionor similar types can’t be Pydantic fields.
Fix 1: Use Depends() Correctly
Every dependency must be wrapped with Depends() in the route function signature:
from fastapi import Depends, FastAPI
from sqlalchemy.orm import Session
from .database import get_db
app = FastAPI()
# WRONG — FastAPI treats 'db' as a query parameter, not a dependency
@app.get("/users")
def get_users(db: Session): # ← Missing Depends()
return db.query(User).all()
# Results in 422: field required (query param 'db' not provided)
# CORRECT — Depends() tells FastAPI to call get_db() and inject the result
@app.get("/users")
def get_users(db: Session = Depends(get_db)):
return db.query(User).all()Common dependency patterns:
from fastapi import Depends, Header, HTTPException
from fastapi.security import OAuth2PasswordBearer
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="auth/token")
# Dependency from header
async def get_api_key(x_api_key: str = Header(...)):
if x_api_key != settings.API_KEY:
raise HTTPException(status_code=403, detail="Invalid API key")
return x_api_key
# Chained dependency — get_current_user depends on oauth2_scheme
async def get_current_user(token: str = Depends(oauth2_scheme)) -> User:
user = await verify_token(token)
if not user:
raise HTTPException(status_code=401, detail="Invalid token")
return user
# Route using multiple dependencies
@app.get("/profile")
async def get_profile(
current_user: User = Depends(get_current_user),
db: Session = Depends(get_db),
):
return db.query(Profile).filter_by(user_id=current_user.id).first()Fix 2: Write Database Session Dependencies Correctly
The database session dependency must use yield (generator pattern) to guarantee cleanup even when exceptions occur:
# WRONG — no cleanup guarantee
def get_db_wrong():
db = SessionLocal()
return db # If a route raises, db.close() never called → connection leak
# WRONG — try/finally but sync in async app
def get_db_sync():
db = SessionLocal()
try:
yield db
finally:
db.close()
# Works, but blocks the event loop in async FastAPI apps
# CORRECT — sync SQLAlchemy with sync dependency
from database import SessionLocal
def get_db() -> Generator:
db = SessionLocal()
try:
yield db # FastAPI calls everything before yield, then route handler runs
db.commit() # Commit on success
except Exception:
db.rollback() # Rollback on any exception
raise
finally:
db.close() # Always runs — even if route raises an exception
# CORRECT — async SQLAlchemy with async dependency
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
async_session = async_sessionmaker(engine, expire_on_commit=False)
async def get_async_db() -> AsyncGenerator[AsyncSession, None]:
async with async_session() as session:
try:
yield session
await session.commit()
except Exception:
await session.rollback()
raise
# Session closed automatically by async context managerPro Tip: Using
async with sessionas a context manager in the async dependency automatically handles session lifecycle. Theyieldinside theasync withblock means FastAPI’s cleanup runs before theasync withexits, ensuring the session commits or rolls back before closing.
Fix 3: Use Class-Based Dependencies
Class-based dependencies are useful when a dependency needs configuration or state:
from fastapi import Depends
class PaginationParams:
def __init__(self, page: int = 1, page_size: int = 20):
if page_size > 100:
raise HTTPException(status_code=400, detail="page_size max is 100")
self.page = page
self.page_size = page_size
self.offset = (page - 1) * page_size
# FastAPI calls PaginationParams(page=..., page_size=...) from query params
@app.get("/items")
def list_items(pagination: PaginationParams = Depends(PaginationParams)):
return db.query(Item).offset(pagination.offset).limit(pagination.page_size).all()Configurable dependency with __call__:
class RateLimiter:
def __init__(self, max_requests: int, window_seconds: int):
self.max_requests = max_requests
self.window_seconds = window_seconds
async def __call__(self, request: Request):
client_ip = request.client.host
key = f"rate_limit:{client_ip}"
count = await redis.incr(key)
if count == 1:
await redis.expire(key, self.window_seconds)
if count > self.max_requests:
raise HTTPException(status_code=429, detail="Too many requests")
# Create instances with different limits
api_rate_limit = RateLimiter(max_requests=100, window_seconds=60)
login_rate_limit = RateLimiter(max_requests=5, window_seconds=300)
@app.post("/auth/login")
async def login(
credentials: LoginCredentials,
_: None = Depends(login_rate_limit), # Strict limit on login
):
return await auth_service.login(credentials)
@app.get("/api/data")
async def get_data(
_: None = Depends(api_rate_limit),
current_user: User = Depends(get_current_user),
):
return data_service.get_all()Fix 4: Fix Async Dependency Mistakes
FastAPI supports both sync and async dependencies, but mixing them incorrectly causes blocking:
# WRONG — sync dependency that does async work (blocks event loop)
def get_current_user_wrong(token: str = Depends(oauth2_scheme)):
# asyncio.run() inside sync dependency blocks the event loop
user = asyncio.run(verify_token(token)) # Never do this
return user
# CORRECT — async dependency for async work
async def get_current_user(token: str = Depends(oauth2_scheme)) -> User:
user = await verify_token(token) # Properly awaited
if not user:
raise HTTPException(status_code=401, detail="Invalid credentials")
return user
# ALSO CORRECT — sync dependency for sync work (FastAPI runs in thread pool)
def get_settings() -> Settings:
return Settings() # No async work — sync is fineAsync generator dependency:
# WRONG — sync generator for async cleanup
def get_http_client_wrong():
client = httpx.Client()
yield client
client.close() # Sync close in async context
# CORRECT — async generator
async def get_http_client() -> AsyncGenerator[httpx.AsyncClient, None]:
async with httpx.AsyncClient() as client:
yield client
# client closed automatically by async context manager
@app.get("/external-data")
async def get_external(
client: httpx.AsyncClient = Depends(get_http_client)
):
response = await client.get("https://api.example.com/data")
return response.json()Fix 5: Override Dependencies in Tests
FastAPI’s dependency override system replaces dependencies during testing — crucial for swapping real databases with test databases:
# tests/conftest.py
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from app.main import app
from app.database import get_db, Base
# Test database — SQLite in-memory
SQLALCHEMY_TEST_URL = "sqlite:///./test.db"
engine = create_engine(SQLALCHEMY_TEST_URL, connect_args={"check_same_thread": False})
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base.metadata.create_all(bind=engine)
def override_get_db():
db = TestingSessionLocal()
try:
yield db
finally:
db.close()
# Override the dependency
app.dependency_overrides[get_db] = override_get_db
@pytest.fixture
def client():
return TestClient(app)
# Clean up after each test
@pytest.fixture(autouse=True)
def reset_db():
yield
Base.metadata.drop_all(bind=engine)
Base.metadata.create_all(bind=engine)Override authentication dependencies in tests:
from app.dependencies import get_current_user
from app.models import User
# Mock authenticated user for tests
def override_get_current_user():
return User(id=1, email="[email protected]", role="admin")
app.dependency_overrides[get_current_user] = override_get_current_user
def test_protected_route(client):
response = client.get("/admin/users")
assert response.status_code == 200 # Passes without a real JWT tokenRemove overrides after tests to avoid bleeding into other test files:
@pytest.fixture(autouse=True)
def clean_overrides():
yield
app.dependency_overrides.clear() # Reset all overrides after each testFix 6: Use Background Tasks vs Dependencies
Dependencies execute synchronously before the response — don’t use them for fire-and-forget background work:
from fastapi import BackgroundTasks
# WRONG — sending email in a dependency blocks the response
async def send_welcome_email(user: User = Depends(get_current_user)):
await email_service.send(user.email, "Welcome!")
# Route waits for email to send before returning
# CORRECT — use BackgroundTasks for post-response work
@app.post("/users", status_code=201)
async def create_user(
user_data: UserCreate,
background_tasks: BackgroundTasks,
db: Session = Depends(get_db),
):
user = await user_service.create(db, user_data)
# Schedule email — sent after response is returned
background_tasks.add_task(email_service.send_welcome, user.email)
return user # Response returned immediatelyFix 7: Sub-dependencies and Dependency Caching
FastAPI caches dependency results within a request — the same dependency instance is reused if called multiple times:
# get_db is called only ONCE per request, even if multiple dependencies use it
async def get_current_user(db: Session = Depends(get_db)) -> User:
return db.query(User).filter_by(id=token_user_id).first()
async def get_user_permissions(
current_user: User = Depends(get_current_user), # Uses cached get_db
) -> list[str]:
return current_user.permissions
@app.get("/dashboard")
async def dashboard(
current_user: User = Depends(get_current_user), # Cached
permissions: list[str] = Depends(get_user_permissions), # Uses same cached instances
db: Session = Depends(get_db), # Same session instance as above
):
# get_db called once total — not 3 times
return {"user": current_user, "permissions": permissions}Disable caching with use_cache=False — for dependencies that should always re-execute:
@app.get("/sensitive-data")
async def sensitive(
# Re-validates token on every call (doesn't use cached result)
user: User = Depends(get_current_user, use_cache=False),
):
return {"data": "very sensitive"}Still Not Working?
Inspect the dependency graph with FastAPI’s debug tools:
from fastapi import FastAPI
from fastapi.routing import APIRoute
app = FastAPI()
# After adding routes, print all dependencies
for route in app.routes:
if isinstance(route, APIRoute):
print(f"{route.path}: {route.dependencies}")Check for Annotated type hint issues (Python 3.9+):
from typing import Annotated
from fastapi import Depends
# Modern style — Annotated[Type, Depends(fn)]
DbSession = Annotated[Session, Depends(get_db)]
CurrentUser = Annotated[User, Depends(get_current_user)]
@app.get("/users")
async def get_users(db: DbSession, user: CurrentUser):
return db.query(User).all()422 Unprocessable Entity on a dependency parameter — FastAPI may try to read a dependency parameter from the request body/query. Ensure the dependency parameter uses Depends() and not a raw type annotation.
For related FastAPI issues, see Fix: FastAPI 422 Unprocessable Entity and Fix: Pydantic Validation Error.
Solo developer based in Japan. Every solution is cross-referenced with official documentation and tested before publishing.
Was this article helpful?
Related Articles
Fix: FastAPI BackgroundTasks Not Working — Task Not Running or Dependency Errors
How to fix FastAPI BackgroundTasks — task not executing, dependency injection in tasks, error handling, Celery for heavy tasks, and lifespan-managed background workers.
Fix: Pydantic ValidationError — Field Required, Value Not a Valid Type, or Extra Fields
How to fix Pydantic v2 validation errors — required fields, type coercion, model_validator, custom validators, extra fields config, and migrating from Pydantic v1.
Fix: Python asyncio Blocking the Event Loop — Mixing Sync and Async Code
How to fix Python asyncio event loop blocking — using run_in_executor for sync calls, asyncio.to_thread, avoiding blocking I/O in coroutines, and detecting event loop stalls.
Fix: Pydantic ValidationError — Field Required / Value Not Valid
How to fix Pydantic ValidationError in Python — missing required fields, wrong types, custom validators, handling optional fields, v1 vs v2 API differences, and debugging complex nested models.