Dependency Injection with Depends
22 min · Intermediate
Need help? Ask a tutor on WhatsApp (opens in a new tab)Routes often need the same helpers: paging values, the current user, a database session. Dependency injection lets a route ask for them in its signature, and FastAPI builds them for each request.
Depends hands a route what it needs
Dependency injection means a function does not create what it needs. It declares it, and the framework creates it and passes it in. In FastAPI the declaration is Depends(function).
Many routes need the same thing: paging values, the current user, a database session. Write that once as a function and ask for it with Depends. FastAPI calls the function for every request and passes the result to your route. This is called dependency injection.
from fastapi import Depends
def paging(skip: int = 0, limit: int = 10): return {"skip": skip, "limit": limit}
@app.get("/orders")def list_orders(page: dict = Depends(paging)): return pageExample 1 · Shared paging values
from fastapi import FastAPIfrom fastapi.testclient import TestClient
app = FastAPI()from fastapi import Depends
def paging(skip: int = 0, limit: int = 10): return {"skip": skip, "limit": min(limit, 50)}
@app.get("/orders")def list_orders(page: dict = Depends(paging)): return {"kind": "orders", **page}
@app.get("/customers")def list_customers(page: dict = Depends(paging)): return {"kind": "customers", **page}
c = TestClient(app)print(c.get("/orders").json())print(c.get("/customers?skip=20&limit=500").json())print(c.get("/orders?limit=abc").status_code){'kind': 'orders', 'skip': 0, 'limit': 10}
{'kind': 'customers', 'skip': 20, 'limit': 50}
422What this code does
- Two routes share one dependency, so the rules for paging are written once.
- The dependency caps
limitat 50, and a bad value still gives 422.
The dependency has its own parameters, so skip and limit become query parameters of the route, with the same validation and the same docs.
yield: set up, use, clean up
A dependency can use yield instead of return. The code before yield runs before the route. The value after yield goes to the route. The code after yield (in finally) runs when the request is done, even after an error. This is the right place to open and close a database session.
def get_db(): db = SessionLocal() try: yield db finally: db.close()
@app.get("/products")def list_products(db: Session = Depends(get_db)): ...Example 2 · Cleanup after yield, even after an error
from fastapi import FastAPIfrom fastapi.testclient import TestClient
app = FastAPI()from fastapi import Depends, HTTPException
log = []
def get_resource(): log.append("open") try: yield "connection" finally: log.append("close")
@app.get("/ok")def ok(res: str = Depends(get_resource)): log.append("route ok") return {"res": res}
@app.get("/fail")def fail(res: str = Depends(get_resource)): log.append("route fail") raise HTTPException(status_code=404, detail="missing")
c = TestClient(app)print(c.get("/ok").status_code, log)log.clear()print(c.get("/fail").status_code, log)200 ['open', 'route ok', 'close'] 404 ['open', 'route fail', 'close']
What this code does
- The order is always open, route, close.
- The
finallyblock runs after the 404 too, so the resource is not leaked.
Example 3 · A database session per request
from fastapi import Depends, FastAPIfrom fastapi.testclient import TestClientfrom pydantic import BaseModelfrom sqlalchemy import create_engine, selectfrom sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column, sessionmakerfrom sqlalchemy.pool import StaticPool
engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool)SessionLocal = sessionmaker(bind=engine)
class Base(DeclarativeBase): pass
class Product(Base): __tablename__ = "products" id: Mapped[int] = mapped_column(primary_key=True) name: Mapped[str] price: Mapped[float]
Base.metadata.create_all(engine)
def get_db(): db = SessionLocal() try: yield db finally: db.close()
class ProductIn(BaseModel): name: str price: float
app = FastAPI()
@app.post("/products", status_code=201)def add_product(body: ProductIn, db: Session = Depends(get_db)): item = Product(name=body.name, price=body.price) db.add(item) db.commit() return {"id": item.id, "name": item.name}
@app.get("/products")def list_products(db: Session = Depends(get_db)): return [p.name for p in db.scalars(select(Product).order_by(Product.id))]
client = TestClient(app)print(client.post("/products", json={"name": "Keyboard", "price": 49.5}).json())client.post("/products", json={"name": "Monitor", "price": 180})print(client.get("/products").json()){'id': 1, 'name': 'Keyboard'}
['Keyboard', 'Monitor']What this code does
StaticPoolkeeps one in-memory SQLite connection that every thread can use, which suits tests.- The routes get a session from
get_dband never close it themselves.
Every request gets its own session and the session is always closed. A route never creates or closes sessions itself.
Dependencies can need other dependencies
A dependency may use Depends too. A common chain is: read a header, find the user, then check the role. FastAPI builds the chain in order and calls each function once per request, even if two parameters ask for the same one. You can also attach a dependency to a whole router.
def current_user(x_user: str = Header()): return {"name": x_user}
def admin_only(user=Depends(current_user)): if user["name"] != "admin": raise HTTPException(status_code=403) return userExample 4 · A chain of dependencies
from fastapi import FastAPIfrom fastapi.testclient import TestClient
app = FastAPI()from fastapi import Depends, Header, HTTPException
def current_user(x_user: str = Header()): return {"name": x_user}
def admin_only(user: dict = Depends(current_user)): if user["name"] != "admin": raise HTTPException(status_code=403, detail="Admins only") return user
@app.get("/me")def me(user: dict = Depends(current_user)): return user
@app.delete("/orders/{order_id}")def delete_order(order_id: int, user: dict = Depends(admin_only)): return {"deleted": order_id, "by": user["name"]}
c = TestClient(app)print(c.get("/me", headers={"X-User": "asha"}).json())print(c.get("/me").status_code)r = c.delete("/orders/4", headers={"X-User": "asha"})print(r.status_code, r.json())print(c.delete("/orders/4", headers={"X-User": "admin"}).json()){'name': 'asha'}
422
403 {'detail': 'Admins only'}
{'deleted': 4, 'by': 'admin'}What this code does
admin_onlyusescurrent_user, so one header check serves both routes.- A header named
x_userin Python is read from theX-Userheader.
Failing inside a dependency stops the request before the route runs. A missing header gives 422, and a wrong user gives 403.
Replace dependencies in tests
In a test you do not want the real database or the real clock. app.dependency_overrides is a dict that maps a dependency function to a replacement. While it is set, FastAPI calls the replacement instead. Clear it after the test.
def fake_db(): yield FakeSession()
app.dependency_overrides[get_db] = fake_db# ... run the test ...app.dependency_overrides.clear()Example 5 · Replace a dependency in a test
from fastapi import FastAPIfrom fastapi.testclient import TestClient
app = FastAPI()from fastapi import Depends
class RealRepo: def names(self): return ["from the database"]
class FakeRepo: def names(self): return ["fake A", "fake B"]
def get_repo(): return RealRepo()
@app.get("/customers")def customers(repo=Depends(get_repo)): return repo.names()
c = TestClient(app)print(c.get("/customers").json())
app.dependency_overrides[get_repo] = lambda: FakeRepo()print(c.get("/customers").json())
app.dependency_overrides.clear()print(c.get("/customers").json())['from the database'] ['fake A', 'fake B'] ['from the database']
What this code does
- The override is a plain function with no arguments needed.
- After
clear()the real dependency is used again.
The route code does not change at all. Only the object it receives changes, which is why dependencies make tests easy.
Next step
Check your understanding
12 quick questions