The Complete Production SOP: How to Deploy and Run FastAPI Dependency Injection & Security OAuth2 Schemes in Autonomous AI Agents

---
name: "fastapi-dependency-injection-security"
description: "Build modular, secure FastAPI endpoints using Depends() and OAuth2 password bearer schemes."
metadata:
  author: "Tiangolo"
  version: "1.0.0"
  runtime_compatibility:
    - "codex"
    - "claude-code"
    - "cursor"
    - "windsurf"
triggers:
  - "fastapi dependency injection depends"
  - "oauth2passwordbearer fastapi"
  - "database session dependency fastapi"
permissions:
  fs_read: true
  fs_write: "workspace"
  shell_exec: false
  net_access: false
---

from typing import AsyncGenerator
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
import jwt

SECRET_KEY = "CHANGE_THIS_IN_PROD_USE_ENV_VAR"
ALGORITHM = "HS256"
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/token")
engine = create_async_engine("sqlite+aiosqlite:///./test.db", echo=False)
SessionLocal = async_sessionmaker(autocommit=False, autoflush=False, bind=engine, class_=AsyncSession)

async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with SessionLocal() as session:
        try:
            yield session
        finally:
            await session.close()

async def get_current_user(token: str = Depends(oauth2_scheme), db: AsyncSession = Depends(get_db)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        user_id: str = payload.get("sub")
        if user_id is None:
            raise credentials_exception
    except jwt.PyJWTError:
        raise credentials_exception
    # In production, query user from db: user = await db.get(User, user_id)
    return {"user_id": user_id, "role": payload.get("role", "user")}

from fastapi import APIRouter, Depends
from app.core.deps import get_current_user

router = APIRouter(prefix="/users", tags=["users"])

@router.get("/me")
async def read_current_user(current_user: dict = Depends(get_current_user)):
    """Retrieve the profile of the currently authenticated user."""
    return {"status": "success", "data": current_user}

@router.get("/admin/dashboard")
async def read_admin_dashboard(current_user: dict = Depends(get_current_user)):
    """Endpoint restricted by role evaluation derived from injected user."""
    if current_user.get("role") != "admin":
        from fastapi import HTTPException, status
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Insufficient permissions")
    return {"status": "authorized", "metrics": {"active_nodes": 42}}

import pytest
from httpx import AsyncClient, ASGITransport
from fastapi import FastAPI
from app.core.deps import get_current_user
from app.api.v1.routes import router

app = FastAPI()
app.include_router(router)

@pytest.mark.asyncio
async def test_read_current_user_override():
    mock_user = {"user_id": "usr_mock_123", "role": "admin"}
    app.dependency_overrides[get_current_user] = lambda: mock_user
    
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as client:
        response = await client.get("/users/me")
        assert response.status_code == 200
        assert response.json() == {"status": "success", "data": mock_user}
    
    app.dependency_overrides.clear()