FastAPI Dependency Injection for Multi-Tenant Request Context: Avoiding the Global State Trap

I’ve built 9 AI features into CitizenApp across 3 different tenant architectures. The worst decision I made? Trying to pass tenant ID through middleware and global context variables. The best? Leaning entirely on FastAPI’s dependency injection system.

Most developers reach for middleware or contextvars for tenant isolation. Both feel natural—they’re “global” enough to avoid passing arguments everywhere, but scoped to the request. I prefer neither. Here’s why: dependencies are explicit, traceable, and testable. When I read a handler signature, I know exactly what data it needs. When a test fails, I’m not debugging some RequestContext that was mutated somewhere in the call stack.

This post shows you how to build type-safe, request-scoped tenant isolation using FastAPI’s native dependency graph—no global variables, no context var debugging nightmares, no middleware pollution.

The Problem with Middleware and Context Vars

Let me show you what I used to do, and why it burned me.

# ❌ The global context var approach (this is what I did first)
from contextvars import ContextVar

tenant_id: ContextVar[str] = ContextVar("tenant_id", default=None)
current_user: ContextVar[dict] = ContextVar("current_user", default=None)

@app.middleware("http")
async def extract_tenant(request: Request, call_next):
    token = request.headers.get("authorization", "").replace("Bearer ", "")
    payload = jwt.decode(token, SECRET)
    
    tenant_id.set(payload["tenant_id"])
    current_user.set(payload["user"])
    
    response = await call_next(request)
    return response

# Later, in a handler
@app.get("/features")
async def list_features():
    tid = tenant_id.get()  # Where did this come from? Good luck debugging.
    user = current_user.get()
    # ...

Problems:

1. Hidden dependencies. The handler doesn’t declare what it needs. A teammate reads list_features() and has no idea it relies on middleware magic.
2. Testing nightmare. To test this endpoint, you need to either mock context vars (brittle) or hit the middleware (slow, fragile).
3. Type safety is gone. tenant_id.get() returns str | None. You’ll add a runtime check. Copy it 50 times. Miss it once in production.
4. Async context var leaks. If you spawn background tasks or use asyncio.create_task(), context vars don’t propagate. You’ll waste a day on this.

I’ve shipped code like this. It works until it doesn’t.

The Right Way: Dependencies as the Source of Truth

FastAPI’s dependency system is designed for exactly this: request-scoped data that multiple handlers need, with full type safety and testability.

Here’s the pattern:

# schemas.py
from pydantic import BaseModel

class TenantContext(BaseModel):
    tenant_id: str
    user_id: str
    permissions: set[str]

# dependencies.py
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPCredential
import jwt

security = HTTPBearer()

async def get_tenant_context(
    credential: HTTPCredential = Depends(security),
) -> TenantContext:
    """Extract and validate JWT, return typed tenant context."""
    try:
        payload = jwt.decode(
            credential.credentials,
            SECRET,
            algorithms=["HS256"],
        )
    except jwt.InvalidTokenError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid token",
        )
    
    return TenantContext(
        tenant_id=payload["tenant_id"],
        user_id=payload["user_id"],
        permissions=set(payload.get("permissions", [])),
    )

# handlers.py
@app.get("/features")
async def list_features(
    context: TenantContext = Depends(get_tenant_context),
) -> list[Feature]:
    """List features for this tenant. Dependency is explicit in signature."""
    features = await db.query(Feature).where(
        Feature.tenant_id == context.tenant_id
    ).all()
    return features

Why this is better:

1. Explicit dependencies. I see TenantContext = Depends(get_tenant_context) and instantly know what this handler needs.
2. Type safety. context is TenantContext, not dict | None. IDE autocomplete works. Runtime errors become type errors.
3. Testable. I can construct a TenantContext in tests without touching middleware:

# test_features.py
async def test_list_features():
    mock_context = TenantContext(
        tenant_id="test-tenant",
        user_id="test-user",
        permissions={"features:read"},
    )
    
    # Override the dependency for this test
    app.dependency_overrides[get_tenant_context] = lambda: mock_context
    
    async with AsyncClient(app=app, base_url="http://test") as client:
        response = await client.get("/features")
    
    assert response.status_code == 200
    app.dependency_overrides.clear()

4. Traceable. When something goes wrong, I follow the handler → dependency → JWT decode → error. No mystery mutations.

Composing Permissions and DB Access

Real multi-tenant systems need more: row-level security, permission checks, database scoping.

# dependencies.py
async def get_db_session(
    context: TenantContext = Depends(get_tenant_context),
) -> AsyncSession:
    """Return a database session scoped to the tenant."""
    async with AsyncSessionLocal() as session:
        # Add a filter to all queries for this session
        @event.listens_for(session.sync_session, "before_execute", propagate=True)
        def receive_before_execute(conn, clauseelement, multiparams, params, execution_options):
            if isinstance(clauseelement, Select):
                # SQLAlchemy magic to auto-filter by tenant_id
                clauseelement = clauseelement.where(
                    getattr(clauseelement.froms[0].c, "tenant_id") == context.tenant_id
                )
        
        yield session

async def require_permission(
    required: set[str],
    context: TenantContext = Depends(get_tenant_context),
) -> TenantContext:
    """Dependency factory: require specific permissions."""
    if not required.issubset(context.permissions):
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Insufficient permissions",
        )
    return context

# handlers.py
@app.post("/features")
async def create_feature(
    payload: CreateFeatureRequest,
    context: TenantContext = Depends(require_permission({"features:write"})),
    session: AsyncSession = Depends(get_db_session),
) -> Feature:
    """Create a feature. Permissions checked, tenant scoped, automatic RLS."""
    feature = Feature(
        tenant_id=context.tenant_id,
        name=payload.name,
        created_by=context.user_id,
    )
    session.add(feature)
    await session.commit()
    return feature

This scales beautifully. Every handler is self-documenting. A new developer reads the signature and knows: “I need a tenant context, and I need write permissions.” Testing? Inject mocks. Debugging? Follow the dependency chain.

Gotcha: Circular Dependencies and Caching

FastAPI dependencies are cached per request by default. This is good (no re-parsing the JWT 5 times). But it can bite you.

# ❌ This creates a circular dependency
async def get_db_session(
    context: TenantContext = Depends(get_tenant_context),
):
    # ...
    yield session

async def get_tenant_context(
    session: AsyncSession = Depends(get_db_session),
):
    # Get tenant metadata from DB
    # CIRCULAR DEPENDENCY ERROR

Solution: Split into smaller dependencies:

# ✅ Correct
async def get_jwt_payload(
    credential: HTTPCredential = Depends(security),
) -> dict:
    """Just decode the JWT."""
    return jwt.decode(credential.credentials, SECRET, algorithms=["HS256"])

async def get_tenant_context(
    payload: dict = Depends(get_jwt_payload),
) -> TenantContext:
    """Build context from payload."""
    return TenantContext(
        tenant_id=payload["tenant_id"],
        user_id=payload["user_id"],
        permissions=set(payload.get("permissions", [])),
    )

async def get_db_session(
    context: TenantContext = Depends(get_tenant_context),
):
    """DB access depends on context, not vice versa."""
    # ...

What I Missed

The one thing I wish I’d known earlier: use Depends with a callable class for complex, stateful dependencies:

class TenantDB:
    def __init__(self, context: TenantContext, session: AsyncSession):
        self.context = context
        self.session = session
    
    async def get_features(self) -> list[Feature]:
        return await self.session.query(Feature).where(
            Feature.tenant_id == self.context.tenant_id
        ).all()

async def get_tenant_db(
    context: TenantContext = Depends(get_tenant_context),
    session: AsyncSession = Depends(get_db_session),
) -> TenantDB:
    return TenantDB(context, session)

@app.get("/features")
async def list_features(db: TenantDB = Depends(get_tenant_db)):
    return await db.get_features()

This abstracts database access and keeps handlers