FastAPI Dependencies and Dependency Injection - Interview Questions
1. What are dependencies in FastAPI and why are they used?
Dependencies in FastAPI are reusable components that can be shared across multiple endpoints. They’re used for:
- Code Reuse: Avoid duplicating common logic
- Authentication: Verify user credentials
- Database Connections: Manage database sessions
- Configuration: Share settings across endpoints
- Validation: Common validation logic
- Caching: Shared cache instances
from fastapi import Depends, FastAPI
app = FastAPI()
def get_db():
db = Database()
try:
yield db
finally:
db.close()
@app.get("/users")
def get_users(db: Database = Depends(get_db)):
return db.get_users()
2. How do you create a simple dependency function?
from fastapi import Depends, FastAPI, HTTPException
from typing import Optional
app = FastAPI()
def get_current_user(token: str = Header(None)):
if not token:
raise HTTPException(status_code=401, detail="Token required")
# Validate token logic here
return {"user_id": 123, "username": "john"}
@app.get("/profile")
def get_profile(current_user: dict = Depends(get_current_user)):
return {"message": f"Hello {current_user['username']}"}
3. What is the difference between function dependencies and class dependencies?
Function Dependencies:
def get_db():
db = Database()
try:
yield db
finally:
db.close()
@app.get("/users")
def get_users(db: Database = Depends(get_db)):
return db.get_users()
Class Dependencies:
class Database:
def __init__(self):
self.connection = create_connection()
def get_users(self):
return self.connection.query("SELECT * FROM users")
def close(self):
self.connection.close()
def get_db():
db = Database()
try:
yield db
finally:
db.close()
4. How do you use dependencies with parameters?
from fastapi import Depends, FastAPI, Query
from typing import Optional
app = FastAPI()
def get_items(skip: int = Query(0), limit: int = Query(10)):
return {"skip": skip, "limit": limit}
@app.get("/items")
def read_items(items_params: dict = Depends(get_items)):
return items_params
# Alternative approach
def get_items_with_params(skip: int = 0, limit: int = 10):
return {"skip": skip, "limit": limit}
@app.get("/items2")
def read_items2(
skip: int = Query(0),
limit: int = Query(10),
items_params: dict = Depends(get_items_with_params)
):
return items_params
5. How do you create dependencies that depend on other dependencies?
from fastapi import Depends, FastAPI, HTTPException, Header
from typing import Optional
app = FastAPI()
def get_db():
db = Database()
try:
yield db
finally:
db.close()
def get_current_user(token: str = Header(None), db: Database = Depends(get_db)):
if not token:
raise HTTPException(status_code=401, detail="Token required")
# Use db to validate token
user = db.get_user_by_token(token)
if not user:
raise HTTPException(status_code=401, detail="Invalid token")
return user
def get_current_active_user(current_user: User = Depends(get_current_user)):
if not current_user.is_active:
raise HTTPException(status_code=400, detail="Inactive user")
return current_user
@app.get("/profile")
def get_profile(current_user: User = Depends(get_current_active_user)):
return {"user": current_user}
6. What are global dependencies and how do you use them?
Global dependencies are applied to all endpoints in the application:
from fastapi import FastAPI, Depends
from contextlib import asynccontextmanager
@asynccontextmanager
async def lifespan(app: FastAPI):
# Startup
print("Starting up...")
yield
# Shutdown
print("Shutting down...")
app = FastAPI(lifespan=lifespan)
# Global dependency for all endpoints
async def verify_api_key(x_api_key: str = Header(None)):
if x_api_key != "secret-key":
raise HTTPException(status_code=401, detail="Invalid API key")
# Apply to all endpoints
app.dependency_overrides[verify_api_key] = verify_api_key
@app.get("/items")
def get_items():
return {"items": []}
7. How do you handle database dependencies with SQLAlchemy?
from fastapi import Depends, FastAPI
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
# Dependency
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
# Models
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
name = Column(String, index=True)
email = Column(String, unique=True, index=True)
# Endpoints
@app.post("/users")
def create_user(user: UserCreate, db: Session = Depends(get_db)):
db_user = User(name=user.name, email=user.email)
db.add(db_user)
db.commit()
db.refresh(db_user)
return db_user
@app.get("/users")
def get_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
users = db.query(User).offset(skip).limit(limit).all()
return users
8. How do you create authentication dependencies?
from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from jose import JWTError, jwt
from passlib.context import CryptContext
app = FastAPI()
security = HTTPBearer()
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
# Secret key and algorithm
SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
def verify_password(plain_password, hashed_password):
return pwd_context.verify(plain_password, hashed_password)
def get_password_hash(password):
return pwd_context.hash(password)
def create_access_token(data: dict):
to_encode = data.copy()
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
return encoded_jwt
async def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(security)):
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(credentials.credentials, SECRET_KEY, algorithms=[ALGORITHM])
username: str = payload.get("sub")
if username is None:
raise credentials_exception
except JWTError:
raise credentials_exception
user = get_user(username)
if user is None:
raise credentials_exception
return user
@app.get("/users/me")
async def read_users_me(current_user: User = Depends(get_current_user)):
return current_user
9. How do you use dependencies for caching?
from fastapi import Depends, FastAPI
from functools import lru_cache
import redis
app = FastAPI()
# Redis cache dependency
def get_redis():
redis_client = redis.Redis(host='localhost', port=6379, db=0)
try:
yield redis_client
finally:
redis_client.close()
# In-memory cache dependency
@lru_cache()
def get_settings():
return Settings()
# Using cache in endpoints
@app.get("/expensive-data")
def get_expensive_data(redis_client: redis.Redis = Depends(get_redis)):
# Check cache first
cached_data = redis_client.get("expensive_data")
if cached_data:
return {"data": cached_data, "source": "cache"}
# If not in cache, compute and store
data = compute_expensive_data()
redis_client.setex("expensive_data", 3600, data) # Cache for 1 hour
return {"data": data, "source": "computed"}
@app.get("/settings")
def get_app_settings(settings: Settings = Depends(get_settings)):
return settings
10. How do you create dependencies for external services?
from fastapi import Depends, FastAPI
import httpx
import asyncio
app = FastAPI()
# HTTP client dependency
async def get_http_client():
async with httpx.AsyncClient() as client:
yield client
# External API service dependency
class ExternalAPIService:
def __init__(self, client: httpx.AsyncClient):
self.client = client
self.base_url = "https://api.external.com"
async def get_user_data(self, user_id: int):
response = await self.client.get(f"{self.base_url}/users/{user_id}")
return response.json()
def get_external_api(client: httpx.AsyncClient = Depends(get_http_client)):
return ExternalAPIService(client)
@app.get("/users/{user_id}/external-data")
async def get_user_external_data(
user_id: int,
external_api: ExternalAPIService = Depends(get_external_api)
):
data = await external_api.get_user_data(user_id)
return {"user_id": user_id, "external_data": data}
11. How do you handle dependency overrides for testing?
from fastapi import Depends, FastAPI
from fastapi.testclient import TestClient
app = FastAPI()
def get_db():
return RealDatabase()
def get_test_db():
return TestDatabase()
# Override dependency for testing
app.dependency_overrides[get_db] = get_test_db
# Test client
client = TestClient(app)
# In your test file
def test_get_users():
response = client.get("/users")
assert response.status_code == 200
# Reset overrides after tests
def teardown_function():
app.dependency_overrides.clear()
12. How do you create async dependencies?
from fastapi import Depends, FastAPI
import asyncio
import aiohttp
app = FastAPI()
# Async database dependency
async def get_async_db():
db = AsyncDatabase()
try:
await db.connect()
yield db
finally:
await db.close()
# Async cache dependency
async def get_async_cache():
cache = AsyncCache()
try:
await cache.connect()
yield cache
finally:
await cache.close()
# Async external service dependency
async def get_async_http_client():
async with aiohttp.ClientSession() as session:
yield session
@app.get("/users")
async def get_users(
db: AsyncDatabase = Depends(get_async_db),
cache: AsyncCache = Depends(get_async_cache)
):
# Check cache first
cached_users = await cache.get("users")
if cached_users:
return cached_users
# Get from database
users = await db.get_users()
await cache.set("users", users, expire=3600)
return users
Do you need a DI container?
Usually no. Depends() already gives you constructor-style wiring with per-request caching, and it’s the idiom the framework and its ecosystem assume.
Reach for a container (dependency-injector, punq, or a hand-rolled composition root) when:
- The dependency graph is genuinely deep — repositories, use cases, external clients — and you want one place to declare lifetimes (singleton / request-scoped / transient).
- You want interface-based wiring: “when something asks for
PaymentGateway, give itStripeGateway”, swappable per environment without touching route signatures. - Several implementations of one abstraction are selected by config (cache backends, queue backends).
- Team convention already uses one.
The pragmatic path: start with Depends() only. When the boilerplate repeats or wiring starts leaking into route code, introduce a composition root that builds the object graph once and hands it to routes — then app.dependency_overrides remains your test seam either way.
See also ../../13_architecture_design/07_di_pattern_library.md and ../../13_architecture_design/03_dependency_injection.md.
Interview angle
- “How does
Depends()work?” - FastAPI inspects the signature, resolves each dependency (recursively), and caches the result per request. A dependency usingyieldruns setup before the handler and teardown after the response, which is how sessions and connections are scoped. - “How do you override a dependency in tests?” -
app.dependency_overrides[get_db] = get_test_db. It’s the cleanest test seam in FastAPI, and it’s why dependencies should be functions rather than module-level globals. - “Where do dependency exceptions get handled?” - a dependency raising
HTTPExceptionshort-circuits before the handler runs, which is exactly how auth and permission checks are expressed. - “When do you need something beyond
Depends()?” - when the same object graph is needed outside HTTP: Celery tasks, CLI commands, Kafka consumers.Depends()is request-scoped and framework-bound, so a container owns the graph and routes consume it. See ../../13_architecture_design/07_di_pattern_library.md.