Module 1: Dependency Injection
Yield Dependencies and Lifecycle — Setup, Cleanup, and Testing
Capsule overview
So far your dependencies do one thing: they run some logic and return a value. But some resources need two steps: setup (preparing the resource) and cleanup (releasing it). A database connection has to open before the endpoint and close after it. A file has to open and close. A transaction has to be committed or rolled back.
Yield dependencies solve this: instead of return, you use yield. Everything before yield is setup (it runs before the endpoint). Everything after yield is cleanup (it runs after the endpoint, even if there was an error). It's the equivalent of a context manager (with), but for dependency injection.
This capsule also introduces dependency_overrides — FastAPI's mechanism for replacing dependencies in testing. If your endpoint uses Depends(get_db), in your tests you can say "when someone asks for get_db, hand them this fake function instead." This is what makes DI fundamental for testing.
The problem: resources that need cleanup
Imagine your app uses a log file:
# ❌ No cleanup — the file can stay open if there's an error
def get_log_file():
log = open("app.log", "a")
return log
@app.post("/tasks")
def create_task(log=Depends(get_log_file)):
log.write("Task created\n")
# If there's an error here, the file never closes
return {"message": "created"}Or imagine a database connection:
# ❌ No cleanup — the session can stay open
def get_db():
db = SessionLocal()
return db
@app.get("/tasks")
def list_tasks(db=Depends(get_db)):
tasks = db.query(Task).all()
# If there's an error, db never closes
# If there's no error, db doesn't close either
return tasksIn both cases, the resource gets created but never released. That causes memory leaks, zombie database connections, and locked files.
Yield dependencies: setup + cleanup
With yield, you split the dependency into two parts:
def get_log_file():
# SETUP: before yield (runs BEFORE the endpoint)
log = open("app.log", "a")
yield log
# CLEANUP: after yield (runs AFTER the endpoint)
log.close()The full flow
Request arrives
↓
SETUP: log = open("app.log", "a")
↓
yield log → the value of log gets passed to the endpoint
↓
The ENDPOINT runs: log.write("Task created\n")
↓
CLEANUP: log.close() (always runs, even if the endpoint raised an error)
↓
Response gets sent
A complete example with a file
import os
from datetime import datetime
from fastapi import FastAPI, Depends
app = FastAPI()
def get_log_writer():
os.makedirs("logs", exist_ok=True)
log_file = open("logs/app.log", "a")
print(f"[SETUP] Log file opened")
yield log_file
log_file.close()
print(f"[CLEANUP] Log file closed")
@app.post("/tasks")
def create_task(log=Depends(get_log_writer)):
timestamp = datetime.now().isoformat()
log.write(f"[{timestamp}] Task created\n")
log.flush()
return {"message": "Task created"}
@app.get("/health")
def health(log=Depends(get_log_writer)):
timestamp = datetime.now().isoformat()
log.write(f"[{timestamp}] Health check\n")
log.flush()
return {"status": "healthy"}Every request opens the file, uses it, and closes it. Guaranteed.
try/finally for safe cleanup
If the cleanup has to happen always — even when there are exceptions — use try/finally:
def get_db_session():
session = create_session()
try:
yield session
finally:
session.close()Simulating a database session
from fastapi import FastAPI, Depends, HTTPException
app = FastAPI()
class FakeDBSession:
def __init__(self):
self.data = [
{"id": 1, "title": "Task 1", "completed": False},
{"id": 2, "title": "Task 2", "completed": True},
{"id": 3, "title": "Task 3", "completed": False},
]
self.closed = False
print(f" [DB] Session opened")
def query_all(self):
return self.data[:]
def query_by_id(self, item_id: int):
for item in self.data:
if item["id"] == item_id:
return item
return None
def add(self, item: dict):
self.data.append(item)
def close(self):
self.closed = True
print(f" [DB] Session closed")
def get_db():
db = FakeDBSession()
try:
yield db
finally:
db.close()
@app.get("/tasks")
def list_tasks(db: FakeDBSession = Depends(get_db)):
return db.query_all()
@app.get("/tasks/{task_id}")
def get_task(task_id: int, db: FakeDBSession = Depends(get_db)):
task = db.query_by_id(task_id)
if not task:
raise HTTPException(status_code=404, detail="Task not found")
return task
@app.post("/tasks", status_code=201)
def create_task(db: FakeDBSession = Depends(get_db)):
new_id = max(t["id"] for t in db.query_all()) + 1
new_task = {"id": new_id, "title": f"Task {new_id}", "completed": False}
db.add(new_task)
return new_taskWhat happens on each request
GET /tasks
[DB] Session opened
→ query_all() returns the list
[DB] Session closed ← cleanup always happens
GET /tasks/999
[DB] Session opened
→ query_by_id(999) returns None
→ HTTPException 404
[DB] Session closed ← cleanup happens even with an error
The finally guarantees that db.close() runs no matter what happens in the endpoint.
Yield with transactions
An advanced pattern is committing or rolling back based on whether there was an error:
from fastapi import FastAPI, Depends, HTTPException
app = FastAPI()
class FakeTransaction:
def __init__(self):
self.operations = []
self.committed = False
self.rolled_back = False
def add_operation(self, op: str):
self.operations.append(op)
def commit(self):
self.committed = True
print(f" [TX] Committed: {self.operations}")
def rollback(self):
self.rolled_back = True
print(f" [TX] Rolled back: {self.operations}")
def get_transaction():
tx = FakeTransaction()
print(" [TX] Transaction started")
try:
yield tx
tx.commit()
except Exception:
tx.rollback()
raise
@app.post("/transfer")
def transfer(tx: FakeTransaction = Depends(get_transaction)):
tx.add_operation("debit account A")
tx.add_operation("credit account B")
return {"status": "success", "operations": tx.operations}
@app.post("/transfer-fail")
def transfer_fail(tx: FakeTransaction = Depends(get_transaction)):
tx.add_operation("debit account A")
raise HTTPException(status_code=500, detail="Network error during credit")curl -X POST http://127.0.0.1:8000/transfer
# [TX] Transaction started
# [TX] Committed: ['debit account A', 'credit account B']
# {"status":"success",...}
curl -X POST http://127.0.0.1:8000/transfer-fail
# [TX] Transaction started
# [TX] Rolled back: ['debit account A']
# {"detail":"Network error during credit"}If the endpoint completes without an error → commit. If it raises an exception → rollback. This is exactly how transactions work with SQLAlchemy in production.
Yield dependencies with async
If you need to do asynchronous cleanup, use async def with yield:
from fastapi import FastAPI, Depends
app = FastAPI()
async def get_async_resource():
print(" [ASYNC] Resource acquired")
resource = {"connection": "active", "pool": "available"}
try:
yield resource
finally:
print(" [ASYNC] Resource released")
@app.get("/data")
async def get_data(resource: dict = Depends(get_async_resource)):
return {"resource_status": resource["connection"]}The async version works just like the sync one, but it lets you use await in the setup and cleanup.
dependency_overrides: replacing dependencies for testing
dependency_overrides is a dictionary on the FastAPI app that lets you replace any dependency with another function. This is essential for testing:
The concept
from fastapi import FastAPI, Depends
app = FastAPI()
def get_db():
return {"type": "production", "url": "postgresql://..."}
@app.get("/info")
def get_info(db: dict = Depends(get_db)):
return {"db_type": db["type"]}In production, get_db returns the real connection. In testing:
def fake_get_db():
return {"type": "test", "url": "sqlite:///:memory:"}
app.dependency_overrides[get_db] = fake_get_dbNow when someone calls an endpoint that uses Depends(get_db), FastAPI runs fake_get_db instead.
A complete example with testing
from fastapi import FastAPI, Depends, HTTPException
from fastapi.testclient import TestClient
app = FastAPI()
production_tasks = [
{"id": 1, "title": "Production Task 1"},
{"id": 2, "title": "Production Task 2"},
]
def get_task_store() -> list:
return production_tasks
def get_task_or_404(task_id: int, store: list = Depends(get_task_store)) -> dict:
for task in store:
if task["id"] == task_id:
return task
raise HTTPException(status_code=404, detail=f"Task {task_id} not found")
@app.get("/tasks")
def list_tasks(store: list = Depends(get_task_store)):
return store
@app.get("/tasks/{task_id}")
def get_task(task: dict = Depends(get_task_or_404)):
return task
# --- Testing ---
test_tasks = [
{"id": 100, "title": "Test Task A"},
{"id": 200, "title": "Test Task B"},
]
def fake_get_task_store() -> list:
return test_tasks
app.dependency_overrides[get_task_store] = fake_get_task_store
client = TestClient(app)
response = client.get("/tasks")
print(response.json())
# [{"id": 100, "title": "Test Task A"}, {"id": 200, "title": "Test Task B"}]
# It uses the test data, not production's!
response = client.get("/tasks/100")
print(response.json())
# {"id": 100, "title": "Test Task A"}
response = client.get("/tasks/1")
print(response.status_code) # 404 — task 1 doesn't exist in test_tasks
# Clear the overrides
app.dependency_overrides.clear()Why does it matter?
Without dependency_overrides, to test an endpoint you'd need:
- A real test database
- Test data inserted into it
- Cleanup after every test
With dependency_overrides:
- You replace the data dependency with a function that returns fake data
- You don't need a database
- The tests are fast, isolated, and predictable
The recommended testing pattern
from fastapi.testclient import TestClient
def test_list_tasks():
test_data = [{"id": 1, "title": "Test"}]
def override_store():
return test_data
app.dependency_overrides[get_task_store] = override_store
client = TestClient(app)
response = client.get("/tasks")
assert response.status_code == 200
assert len(response.json()) == 1
app.dependency_overrides.clear()
def test_task_not_found():
def override_store():
return []
app.dependency_overrides[get_task_store] = override_store
client = TestClient(app)
response = client.get("/tasks/1")
assert response.status_code == 404
app.dependency_overrides.clear()Each test sets up its own override and clears it at the end. That keeps the tests isolated from each other.
App-level dependencies
You can declare dependencies that apply to every endpoint in the app:
from fastapi import FastAPI, Depends, Header, HTTPException
def verify_internal_token(x_internal_token: str = Header(...)):
if x_internal_token != "internal-secret":
raise HTTPException(status_code=403, detail="Not an internal service")
app = FastAPI(dependencies=[Depends(verify_internal_token)])
@app.get("/tasks")
def list_tasks():
return [{"id": 1, "title": "Task 1"}]
@app.get("/users")
def list_users():
return [{"id": 1, "name": "Alice"}]
@app.get("/health")
def health():
return {"status": "healthy"}Every endpoint requires the X-Internal-Token header. You don't need to add dependencies=[Depends(...)] to each one.
The catch: what about public endpoints?
If you use app-level dependencies, every endpoint requires them — including /health, which should probably be public. The fix is to use router-level dependencies (Module 2), not app-level ones. For now, keep this in mind as a concept.
Yield + Override together
You can override yield dependencies too:
from fastapi import FastAPI, Depends
app = FastAPI()
def get_db():
print(" [PRODUCTION] Opening real DB")
db = {"type": "postgres", "data": [{"id": 1, "title": "Real task"}]}
try:
yield db
finally:
print(" [PRODUCTION] Closing real DB")
@app.get("/tasks")
def list_tasks(db: dict = Depends(get_db)):
return db["data"]
# Override for testing
def fake_get_db():
print(" [TEST] Using fake DB")
db = {"type": "memory", "data": [{"id": 99, "title": "Fake task"}]}
try:
yield db
finally:
print(" [TEST] Fake DB cleaned up")
app.dependency_overrides[get_db] = fake_get_dbThe override can be a yield dependency too, preserving the setup/cleanup pattern.
Visual summary: types of dependencies
Dependency Types:
┌─────────────────────────────┐
│ Regular (return) │
│ def get_data(): │
│ return data │
│ → Setup only, no cleanup │
├─────────────────────────────┤
│ Yield (yield) │
│ def get_resource(): │
│ resource = open(...) │
│ yield resource │
│ resource.close() │
│ → Setup + cleanup │
├─────────────────────────────┤
│ Yield + try/finally │
│ def get_session(): │
│ session = Session() │
│ try: │
│ yield session │
│ finally: │
│ session.close() │
│ → GUARANTEED cleanup │
├─────────────────────────────┤
│ Yield + commit/rollback │
│ def get_tx(): │
│ tx = Transaction() │
│ try: │
│ yield tx │
│ tx.commit() │
│ except: │
│ tx.rollback() │
│ raise │
│ → Auto commit/rollback │
└─────────────────────────────┘
Exercises
Exercise 1: A yield dependency for logging (Easy)
Create a yield dependency request_logger that prints "Request started" before the yield and "Request finished" after it. Return a dict with a timestamp. Use it in two endpoints.
See solution
from datetime import datetime
from fastapi import FastAPI, Depends
app = FastAPI()
def request_logger():
start = datetime.now()
print(f"[{start.isoformat()}] Request started")
yield {"started_at": start.isoformat()}
end = datetime.now()
duration = (end - start).total_seconds()
print(f"[{end.isoformat()}] Request finished ({duration:.3f}s)")
@app.get("/tasks")
def list_tasks(log: dict = Depends(request_logger)):
return {"tasks": [], "request_started": log["started_at"]}
@app.get("/health")
def health(log: dict = Depends(request_logger)):
return {"status": "healthy", "request_started": log["started_at"]}curl http://127.0.0.1:8000/tasks
# In the console:
# [2026-03-13T10:00:00] Request started
# [2026-03-13T10:00:00] Request finished (0.001s)Exercise 2: Yield with try/finally (Easy)
Create FakeConnection (a class with open(), close(), is_open). Create a yield dependency that opens the connection, yields it, and closes it in finally. Check that it closes even when the endpoint raises an HTTPException.
See solution
from fastapi import FastAPI, Depends, HTTPException
app = FastAPI()
class FakeConnection:
def __init__(self):
self.is_open = False
def open(self):
self.is_open = True
print(" [CONN] Connection opened")
def close(self):
self.is_open = False
print(" [CONN] Connection closed")
def query(self, sql: str):
if not self.is_open:
raise RuntimeError("Connection is closed")
return [{"result": f"data from: {sql}"}]
def get_connection():
conn = FakeConnection()
conn.open()
try:
yield conn
finally:
conn.close()
@app.get("/data")
def get_data(conn: FakeConnection = Depends(get_connection)):
results = conn.query("SELECT * FROM tasks")
return results
@app.get("/error")
def get_error(conn: FakeConnection = Depends(get_connection)):
raise HTTPException(status_code=500, detail="Simulated error")curl http://127.0.0.1:8000/data
# [CONN] Connection opened
# [CONN] Connection closed
# [{"result":"data from: SELECT * FROM tasks"}]
curl http://127.0.0.1:8000/error
# [CONN] Connection opened
# [CONN] Connection closed ← it closes despite the error
# {"detail":"Simulated error"}Exercise 3: Yield with commit/rollback (Medium)
Create a SimpleTransaction class with execute(op), commit(), and rollback() methods. Create a yield dependency that commits if everything goes well and rolls back if there's an exception. Create two endpoints: one that completes successfully and one that fails on purpose.
See solution
from fastapi import FastAPI, Depends, HTTPException
app = FastAPI()
class SimpleTransaction:
def __init__(self):
self.operations = []
self.state = "active"
def execute(self, op: str):
self.operations.append(op)
print(f" [TX] Execute: {op}")
def commit(self):
self.state = "committed"
print(f" [TX] COMMIT ({len(self.operations)} operations)")
def rollback(self):
self.state = "rolled_back"
print(f" [TX] ROLLBACK ({len(self.operations)} operations discarded)")
def get_transaction():
tx = SimpleTransaction()
print(" [TX] Transaction started")
try:
yield tx
tx.commit()
except Exception:
tx.rollback()
raise
@app.post("/order")
def create_order(tx: SimpleTransaction = Depends(get_transaction)):
tx.execute("INSERT INTO orders (total) VALUES (100)")
tx.execute("UPDATE inventory SET stock = stock - 1")
tx.execute("INSERT INTO order_items (order_id, product_id)")
return {"status": "order created", "operations": len(tx.operations)}
@app.post("/order-fail")
def create_order_fail(tx: SimpleTransaction = Depends(get_transaction)):
tx.execute("INSERT INTO orders (total) VALUES (100)")
tx.execute("UPDATE inventory SET stock = stock - 1")
raise HTTPException(status_code=402, detail="Payment failed")curl -X POST http://127.0.0.1:8000/order
# [TX] Transaction started
# [TX] Execute: INSERT INTO orders...
# [TX] Execute: UPDATE inventory...
# [TX] Execute: INSERT INTO order_items...
# [TX] COMMIT (3 operations)
curl -X POST http://127.0.0.1:8000/order-fail
# [TX] Transaction started
# [TX] Execute: INSERT INTO orders...
# [TX] Execute: UPDATE inventory...
# [TX] ROLLBACK (2 operations discarded)
# {"detail":"Payment failed"}Exercise 4: dependency_overrides for testing (Medium)
Create an app with get_task_store() that returns a list of "production" tasks. Write a test with TestClient that uses dependency_overrides to inject test data. Check that the endpoint returns the test data, not production's.
See solution
from fastapi import FastAPI, Depends
from fastapi.testclient import TestClient
app = FastAPI()
production_data = [
{"id": 1, "title": "Prod Task 1"},
{"id": 2, "title": "Prod Task 2"},
{"id": 3, "title": "Prod Task 3"},
]
def get_task_store() -> list:
return production_data
@app.get("/tasks")
def list_tasks(store: list = Depends(get_task_store)):
return {"count": len(store), "tasks": store}
# --- Tests ---
def test_list_tasks_with_override():
test_data = [{"id": 99, "title": "Test Only"}]
def fake_store():
return test_data
app.dependency_overrides[get_task_store] = fake_store
client = TestClient(app)
response = client.get("/tasks")
assert response.status_code == 200
body = response.json()
assert body["count"] == 1
assert body["tasks"][0]["id"] == 99
assert body["tasks"][0]["title"] == "Test Only"
app.dependency_overrides.clear()
print("✅ Test passed!")
def test_without_override_uses_production():
app.dependency_overrides.clear()
client = TestClient(app)
response = client.get("/tasks")
body = response.json()
assert body["count"] == 3
assert body["tasks"][0]["title"] == "Prod Task 1"
print("✅ Test passed — production data used!")
if __name__ == "__main__":
test_list_tasks_with_override()
test_without_override_uses_production()python app/test_overrides.py
# ✅ Test passed!
# ✅ Test passed — production data used!Exercise 5: Overriding a yield dependency (Hard)
Create a yield dependency get_db that simulates an expensive database connection (it prints "Heavy DB setup" and "Heavy DB cleanup"). Create a fake_get_db override that uses in-memory data (it prints "Lightweight test DB"). Use TestClient to check that the override works and that the fake's cleanup runs too.
See solution
from fastapi import FastAPI, Depends, HTTPException
from fastapi.testclient import TestClient
app = FastAPI()
class HeavyDB:
def __init__(self):
self.data = [
{"id": 1, "title": "Production data 1"},
{"id": 2, "title": "Production data 2"},
]
def get_all(self):
return self.data
def get_by_id(self, item_id: int):
for item in self.data:
if item["id"] == item_id:
return item
return None
def get_db():
print(" [PROD] Heavy DB setup (connecting to PostgreSQL...)")
db = HeavyDB()
try:
yield db
finally:
print(" [PROD] Heavy DB cleanup (closing pool...)")
@app.get("/items")
def list_items(db: HeavyDB = Depends(get_db)):
return db.get_all()
@app.get("/items/{item_id}")
def get_item(item_id: int, db: HeavyDB = Depends(get_db)):
item = db.get_by_id(item_id)
if not item:
raise HTTPException(status_code=404, detail="Not found")
return item
# --- Test override ---
class LightDB:
def __init__(self):
self.data = [{"id": 100, "title": "Test item"}]
def get_all(self):
return self.data
def get_by_id(self, item_id: int):
for item in self.data:
if item["id"] == item_id:
return item
return None
def fake_get_db():
print(" [TEST] Lightweight test DB setup")
db = LightDB()
try:
yield db
finally:
print(" [TEST] Lightweight test DB cleanup")
def test_with_fake_db():
app.dependency_overrides[get_db] = fake_get_db
client = TestClient(app)
response = client.get("/items")
assert response.status_code == 200
assert len(response.json()) == 1
assert response.json()[0]["id"] == 100
response = client.get("/items/100")
assert response.status_code == 200
response = client.get("/items/1")
assert response.status_code == 404
app.dependency_overrides.clear()
print("✅ All tests passed with fake DB!")
if __name__ == "__main__":
test_with_fake_db()python app/test_yield_override.py
# [TEST] Lightweight test DB setup
# [TEST] Lightweight test DB cleanup
# [TEST] Lightweight test DB setup
# [TEST] Lightweight test DB cleanup
# [TEST] Lightweight test DB setup
# [TEST] Lightweight test DB cleanup
# ✅ All tests passed with fake DB!You never see "[PROD] Heavy DB setup" — the override replaces the dependency completely.
Troubleshooting
Problem 1: The cleanup doesn't run
Cause: You aren't using try/finally. If the code before yield raises an exception, Python jumps straight to the nearest except, not to the code after yield.
# ❌ No try/finally — cleanup isn't guaranteed
def get_resource():
resource = acquire()
yield resource
resource.release() # might not run
# ✅ With try/finally — cleanup guaranteed
def get_resource():
resource = acquire()
try:
yield resource
finally:
resource.release() # always runsProblem 2: "ValueError: generator already closed"
Cause: You're trying to do something with the resource after the generator already finished. This can happen if you store the yielded value and use it outside the request's lifecycle.
Fix: Don't hold references to the yielded resource outside the endpoint's scope.
Problem 3: dependency_overrides doesn't work
Cause: The override's key has to be exactly the same function used in Depends().
# ❌ A different function (even if it's named the same)
def get_db():
return "production"
def another_get_db():
return "production"
app.dependency_overrides[another_get_db] = fake # doesn't work if the endpoint uses get_db
# ✅ The same reference
app.dependency_overrides[get_db] = fakeProblem 4: The override leaks into tests that don't expect it
Cause: You aren't clearing the overrides between tests.
# ❌ The override persists across tests
app.dependency_overrides[get_db] = fake_db
# ✅ Clear it after each test
app.dependency_overrides.clear()Problem 5: A yield dependency with multiple yields
Cause: A yield dependency can only have one yield. Multiple yields cause an error.
# ❌ Multiple yields
def get_resources():
db = connect_db()
yield db
cache = connect_cache()
yield cache # ERROR
# ✅ A single yield with multiple resources
def get_resources():
db = connect_db()
cache = connect_cache()
try:
yield {"db": db, "cache": cache}
finally:
db.close()
cache.close()Summary
- Yield dependencies split the execution: before
yield= setup, after it = cleanup try/finallyguarantees the cleanup always happens, even with exceptionstry/exceptin yield dependencies enables automatic commit/rollbackdependency_overridesreplaces dependencies in testing without touching production code- The override's key must be the same function reference used in
Depends() - Always clear overrides with
app.dependency_overrides.clear()between tests - Yield dependencies can only have one
yield - App-level dependencies:
FastAPI(dependencies=[...])applies to every endpoint - Yield dependencies are the foundation of database sessions in production
Additional resources
- FastAPI - Dependencies with yield — The official yield dependencies documentation
- FastAPI - Testing Dependencies — dependency_overrides for testing
- FastAPI - Testing — TestClient and testing basics
- Python - Generators — How generators work in Python
- FastAPI - Global Dependencies — App-level dependencies
- Python - contextmanager — Context managers in Python (a similar pattern)
What's next?
Next capsule: Project — Refactoring the To-Do API with DI. You'll take your existing API and apply everything you've learned: pagination as a dependency, lookup as a dependency, centralized data access, and cleanup with yield. The result will be an API with the same endpoints but significantly cleaner and more maintainable.