Learnixo

FastAPI for AI Engineers · Lesson 7 of 12

Dependency Injection: Auth, DB, Rate Limiting

What Is Dependency Injection?

Dependency Injection (DI) is a pattern where objects receive the resources they need rather than creating them internally. Instead of:

Python
# Without DI — tightly coupled
@app.post("/chat")
async def chat(req: ChatRequest):
    client = AsyncOpenAI(api_key=os.environ["OPENAI_API_KEY"])  # created every call
    db = await asyncpg.connect(DATABASE_URL)                    # created every call
    user = await db.fetchrow("SELECT * FROM users WHERE id=$1", req.user_id)
    # ...

You declare what you need and the framework provides it:

Python
# With DI — loosely coupled, testable
@app.post("/chat")
async def chat(
    req: ChatRequest,
    client: AsyncOpenAI = Depends(get_openai_client),
    db: asyncpg.Connection = Depends(get_db),
    user: User = Depends(get_current_user),
):
    # client, db, and user are injected — no setup code here
    ...

FastAPI's DI system has two key properties:

  1. Automatic — declare the type + Depends(), FastAPI calls the dependency for you
  2. Cacheable — by default, each dependency is called once per request, not once per injection point

The Depends() Primitive

Depends(callable) tells FastAPI to call callable and pass the result to the parameter. The callable can be:

  • A plain function
  • An async def function
  • A class
  • A generator (for teardown)
  • An async generator (for async teardown)
Python
from fastapi import Depends

# Plain function
def get_settings() -> dict:
    return {"env": "production", "debug": False}

@app.get("/config")
async def config(settings: dict = Depends(get_settings)) -> dict:
    return settings

# Async function
async def get_feature_flags() -> dict:
    # Imagine fetching from Redis
    return {"streaming_enabled": True, "rag_enabled": True}

@app.get("/flags")
async def flags(flags: dict = Depends(get_feature_flags)) -> dict:
    return flags

Generator Dependencies: Setup and Teardown

Use a generator (with yield) when you need to clean up a resource after the request is complete:

Python
import asyncpg
from fastapi import Depends
from typing import AsyncGenerator

DATABASE_URL = "postgresql://user:pass@localhost/mydb"

async def get_db() -> AsyncGenerator[asyncpg.Connection, None]:
    conn = await asyncpg.connect(DATABASE_URL)
    try:
        yield conn          # Connection is available during the request
    finally:
        await conn.close()  # Always closes, even if an exception was raised

@app.get("/users/{user_id}")
async def get_user(
    user_id: int,
    db: asyncpg.Connection = Depends(get_db),
) -> dict:
    row = await db.fetchrow("SELECT id, name, email FROM users WHERE id = $1", user_id)
    if row is None:
        raise HTTPException(status_code=404, detail="User not found")
    return dict(row)

The connection is opened before the handler runs and closed after it returns (or after it raises an exception).

Common Dependencies: DB, Redis, Current User

Database session pool

Python
# dependencies.py
import asyncpg
from fastapi import Depends
from typing import AsyncGenerator

_pool: asyncpg.Pool | None = None

async def get_pool() -> asyncpg.Pool:
    global _pool
    if _pool is None:
        _pool = await asyncpg.create_pool(DATABASE_URL, min_size=2, max_size=20)
    return _pool

async def get_db(pool: asyncpg.Pool = Depends(get_pool)) -> AsyncGenerator[asyncpg.Connection, None]:
    async with pool.acquire() as conn:
        yield conn

Redis client

Python
import redis.asyncio as aioredis

_redis: aioredis.Redis | None = None

async def get_redis() -> aioredis.Redis:
    global _redis
    if _redis is None:
        _redis = aioredis.from_url("redis://localhost:6379", decode_responses=True)
    return _redis

Authenticated user

Python
from fastapi import Depends, HTTPException, Header
from pydantic import BaseModel
import jwt

SECRET_KEY = "change-me-in-production"

class User(BaseModel):
    id: str
    email: str
    roles: list[str]

async def get_current_user(
    authorization: str | None = Header(default=None)
) -> User:
    if not authorization or not authorization.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="Missing or invalid Authorization header")

    token = authorization.removeprefix("Bearer ")
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
    except jwt.ExpiredSignatureError:
        raise HTTPException(status_code=401, detail="Token expired")
    except jwt.InvalidTokenError:
        raise HTTPException(status_code=401, detail="Invalid token")

    return User(
        id=payload["sub"],
        email=payload["email"],
        roles=payload.get("roles", []),
    )

Dependency Chain: auth → user → permissions

Dependencies can depend on other dependencies, forming a chain:

Python
def require_role(required_role: str):
    """Factory that returns a dependency checking for a specific role."""
    async def _check_role(user: User = Depends(get_current_user)) -> User:
        if required_role not in user.roles:
            raise HTTPException(
                status_code=403,
                detail=f"Role '{required_role}' required",
            )
        return user
    return _check_role


# Usage — inject a role-checked user in one line
@app.delete("/drugs/{drug_id}")
async def delete_drug(
    drug_id: int,
    user: User = Depends(require_role("admin")),
    db: asyncpg.Connection = Depends(get_db),
) -> dict:
    await db.execute("DELETE FROM drugs WHERE id = $1", drug_id)
    return {"deleted": drug_id, "by": user.email}

The chain is:

delete_drug
  └── require_role("admin")
        └── get_current_user
              └── Header(authorization)

FastAPI resolves the chain automatically, calling each dependency once per request.

Overriding Dependencies in Tests

This is the killer feature of FastAPI's DI: you can swap any dependency for a mock during testing without touching production code.

Python
# tests/test_chat.py
import pytest
from fastapi.testclient import TestClient
from unittest.mock import AsyncMock, MagicMock
from main import app
from dependencies import get_openai_client, get_current_user
from models import User

# --- Mock OpenAI client ---

def make_mock_openai_response(content: str):
    mock_choice = MagicMock()
    mock_choice.message.content = content
    mock_choice.finish_reason = "stop"

    mock_usage = MagicMock()
    mock_usage.prompt_tokens = 10
    mock_usage.completion_tokens = 20

    mock_response = MagicMock()
    mock_response.choices = [mock_choice]
    mock_response.usage = mock_usage
    mock_response.model = "gpt-4o"

    return mock_response

@pytest.fixture
def mock_openai_client():
    client = MagicMock()
    client.chat.completions.create = AsyncMock(
        return_value=make_mock_openai_response("Ibuprofen reduces inflammation.")
    )
    return client

@pytest.fixture
def mock_user():
    return User(id="test-user-1", email="test@example.com", roles=["user"])

@pytest.fixture
def client(mock_openai_client, mock_user):
    app.dependency_overrides[get_openai_client] = lambda: mock_openai_client
    app.dependency_overrides[get_current_user] = lambda: mock_user
    with TestClient(app) as c:
        yield c
    app.dependency_overrides.clear()

def test_chat_endpoint(client):
    response = client.post("/chat", json={
        "messages": [{"role": "user", "content": "What is ibuprofen?"}]
    })
    assert response.status_code == 200
    data = response.json()
    assert "Ibuprofen" in data["content"]
    assert data["finish_reason"] == "stop"

app.dependency_overrides is a dict mapping the real dependency callable to the override callable. FastAPI uses the override whenever the real dependency would have been called.

Async Dependencies

Dependencies can be async def — FastAPI awaits them in the event loop:

Python
import aiofiles
import json

async def load_config_file() -> dict:
    async with aiofiles.open("config.json", "r") as f:
        contents = await f.read()
    return json.loads(contents)

@app.get("/settings")
async def settings(config: dict = Depends(load_config_file)) -> dict:
    return config

Real Example: Inject OpenAI Client, Swap for Mock in Tests

Here is a complete, production-style setup:

Python
# dependencies.py
import os
from openai import AsyncOpenAI, AsyncAzureOpenAI

_openai_client: AsyncOpenAI | None = None

def get_openai_client() -> AsyncOpenAI:
    """
    Return a cached AsyncOpenAI client.
    In tests, override this with app.dependency_overrides.
    """
    global _openai_client
    if _openai_client is None:
        azure_endpoint = os.environ.get("AZURE_OPENAI_ENDPOINT")
        if azure_endpoint:
            _openai_client = AsyncAzureOpenAI(
                api_key=os.environ["AZURE_OPENAI_API_KEY"],
                azure_endpoint=azure_endpoint,
                api_version="2024-12-01-preview",
            )
        else:
            _openai_client = AsyncOpenAI(
                api_key=os.environ["OPENAI_API_KEY"],
            )
    return _openai_client
Python
# routers/drug_ai.py
from fastapi import APIRouter, Depends, HTTPException
from openai import AsyncOpenAI
from pydantic import BaseModel, Field
from dependencies import get_openai_client, get_current_user, get_db
from models import User
import asyncpg

router = APIRouter(prefix="/drug-ai", tags=["drug-ai"])


class DrugQueryRequest(BaseModel):
    drug_name: str = Field(..., min_length=2, max_length=200)
    question: str = Field(..., min_length=5, max_length=1000)


class DrugQueryResponse(BaseModel):
    drug_name: str
    answer: str
    sources: list[str]


@router.post("/query", response_model=DrugQueryResponse)
async def query_drug_info(
    req: DrugQueryRequest,
    client: AsyncOpenAI = Depends(get_openai_client),
    user: User = Depends(get_current_user),
    db: asyncpg.Connection = Depends(get_db),
) -> DrugQueryResponse:
    # Fetch drug context from database
    drug_row = await db.fetchrow(
        "SELECT name, description, interactions FROM drugs WHERE LOWER(name) = LOWER($1)",
        req.drug_name,
    )

    if drug_row is None:
        raise HTTPException(status_code=404, detail=f"Drug '{req.drug_name}' not found in database.")

    context = f"""
Drug: {drug_row['name']}
Description: {drug_row['description']}
Known interactions: {drug_row['interactions']}
""".strip()

    response = await client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "system",
                "content": (
                    "You are a clinical pharmacology assistant. "
                    "Answer questions about medications using only the provided context. "
                    "If the context does not contain the answer, say so clearly."
                ),
            },
            {
                "role": "user",
                "content": f"Context:\n{context}\n\nQuestion: {req.question}",
            },
        ],
        temperature=0.3,
        max_tokens=512,
    )

    return DrugQueryResponse(
        drug_name=drug_row["name"],
        answer=response.choices[0].message.content or "",
        sources=[f"Internal drug database: {drug_row['name']}"],
    )
Python
# tests/test_drug_ai.py
import pytest
from fastapi.testclient import TestClient
from unittest.mock import AsyncMock, MagicMock, patch
from main import app
from dependencies import get_openai_client, get_current_user, get_db
from models import User

FAKE_DB_DRUG = {
    "name": "Ibuprofen",
    "description": "NSAID used to reduce fever and treat pain or inflammation.",
    "interactions": "May interact with blood thinners such as warfarin.",
}

@pytest.fixture
def mock_db():
    conn = MagicMock()
    conn.fetchrow = AsyncMock(return_value=FAKE_DB_DRUG)
    return conn

@pytest.fixture
def mock_openai():
    choice = MagicMock()
    choice.message.content = "Ibuprofen may increase bleeding risk when combined with warfarin."
    client = MagicMock()
    client.chat.completions.create = AsyncMock(
        return_value=MagicMock(choices=[choice])
    )
    return client

@pytest.fixture
def test_client(mock_db, mock_openai):
    test_user = User(id="u1", email="test@test.com", roles=["user"])
    app.dependency_overrides[get_db] = lambda: mock_db
    app.dependency_overrides[get_openai_client] = lambda: mock_openai
    app.dependency_overrides[get_current_user] = lambda: test_user
    with TestClient(app) as c:
        yield c
    app.dependency_overrides.clear()

def test_drug_query_returns_answer(test_client):
    resp = test_client.post(
        "/drug-ai/query",
        json={"drug_name": "ibuprofen", "question": "Does it interact with warfarin?"},
        headers={"Authorization": "Bearer fake-token"},
    )
    assert resp.status_code == 200
    body = resp.json()
    assert body["drug_name"] == "Ibuprofen"
    assert "warfarin" in body["answer"].lower()

def test_drug_not_found(test_client, mock_db):
    mock_db.fetchrow = AsyncMock(return_value=None)
    resp = test_client.post(
        "/drug-ai/query",
        json={"drug_name": "notadrug", "question": "What is it?"},
        headers={"Authorization": "Bearer fake-token"},
    )
    assert resp.status_code == 404

Dependency Caching

By default, FastAPI calls a dependency once per request and reuses the result for all parameters that declare the same dependency. This means get_db is called once even if three route parameters each Depends(get_db).

To disable caching and call the dependency fresh for each injection point:

Python
from fastapi import Depends

@app.get("/example")
async def example(
    db1: asyncpg.Connection = Depends(get_db, use_cache=False),
    db2: asyncpg.Connection = Depends(get_db, use_cache=False),
) -> dict:
    # db1 and db2 are separate connections
    ...

This is rarely needed — the default caching is usually what you want.

Key Takeaways

  • Depends(callable) injects the return value of callable into your route handler — FastAPI calls it for you
  • Use generator dependencies (yield) for resources that need teardown: database connections, file handles
  • Chain dependencies naturally — a dependency can itself depend on other dependencies
  • app.dependency_overrides swaps real dependencies for mocks in tests — the cleanest testing pattern in FastAPI
  • Async dependencies work natively — async def dependencies are awaited in the event loop
  • Dependencies are cached per request by default — get_db() is called once even if multiple parameters share it

Next lesson: Application Lifespan — loading models and opening connection pools at startup.

Previous

Background Tasks: Async Processing Without Blocking

Next Lesson

Lifespan Events: Startup and Shutdown