FastAPI Dependency Injection — How and Why to Use It

from fastapi import FastAPI, Depends, Query, HTTPException, status
from typing import Annotated

app = FastAPI()


def common_params(
    skip: int = Query(default=0, ge=0, description="Number of records to skip"),
    limit: int = Query(default=100, ge=1, le=1000, description="Maximum records to return"),
) -> dict:
    """
    Standardized pagination parameters for all list endpoints.

    Enforces a hard cap at 1000 to prevent unbounded queries.
    ge=0 on skip prevents negative offsets that some ORMs handle
    unpredictably.

    This function is the single source of truth for pagination logic.
    Changing the cap or default here applies to every endpoint that
    uses it — no find-and-replace required.
    """
    return {"skip": skip, "limit": limit}


# Type alias for readability — used across multiple route files
PaginationParams = Annotated[dict, Depends(common_params)]


@app.get("/users")
def list_users(params: PaginationParams):
    """
    List users with standardized pagination.
    Query params 'skip' and 'limit' appear in OpenAPI docs
    even though they are declared inside common_params, not here.
    """
    # Business logic: fetch users using params["skip"] and params["limit"]
    return {"context": "users", "pagination": params}


@app.get("/orders")
def list_orders(params: PaginationParams):
    """
    List orders with the same pagination contract.
    If the cap changes from 1000 to 500, it changes here
    and applies to both /users and /orders automatically.
    """
    return {"context": "orders", "pagination": params}


@app.get("/products")
def list_products(params: PaginationParams, category: str | None = None):
    """
    Dependencies compose cleanly with route-level parameters.
    category is unique to this endpoint; skip and limit come from the dependency.
    Both appear in the OpenAPI schema.
    """
    return {"context": "products", "category": category, "pagination": params}

from fastapi import FastAPI, Depends, HTTPException, Header, status, APIRouter
from typing import Annotated
from dataclasses import dataclass


@dataclass
class AuthenticatedUser:
    """
    The identity object returned by a successful auth dependency.

    Returning a structured object instead of a raw dict makes
    downstream role checks type-safe and self-documenting.
    Mypy catches accesses to non-existent fields at development time
    rather than at runtime in production.
    """
    user_id: str
    role: str
    scopes: list[str]


def validate_api_key(
    x_forge_token: Annotated[str | None, Header()] = None,
) -> AuthenticatedUser:
    """
    Validate the X-Forge-Token header against the allowed key set.

    In production, replace the hardcoded check with a lookup against
    HashiCorp Vault, AWS Secrets Manager, or a secrets database.
    Never hardcode secrets in source — this example uses a literal
    only to show the dependency structure.

    Raises HTTPException 401 if the token is missing or invalid.
    Returns an AuthenticatedUser if valid — the route receives the
    identity, not a boolean.
    """
    if not x_forge_token or x_forge_token != "forge-prod-secret":
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid or missing Forge API Key",
            headers={"WWW-Authenticate": "ApiKey"},
        )
    # In production: look up the key, fetch the associated user and scopes
    return AuthenticatedUser(
        user_id="svc-order-service",
        role="service",
        scopes=["orders:read", "orders:write"],
    )


def require_admin(
    auth: Annotated[AuthenticatedUser, Depends(validate_api_key)],
) -> AuthenticatedUser:
    """
    Sub-dependency for admin-only endpoints.

    Composes on top of validate_api_key — the request must first
    pass the base auth check, then pass the role check.
    FastAPI caches the validate_api_key result, so the token is
    only validated once even though two dependencies reference it.
    """
    if auth.role != "admin":
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail=f"Admin role required. Caller role: {auth.role}",
        )
    return auth


# Protected router — every route inherits auth at the router level
# No individual route can skip auth by forgetting the Depends() annotation
protected_router = APIRouter(
    prefix="/api/v1",
    dependencies=[Depends(validate_api_key)],
)

# Admin router — sub-dependency composes on the base auth check
admin_router = APIRouter(
    prefix="/admin",
    dependencies=[Depends(require_admin)],
)


@protected_router.get("/orders")
def list_orders(auth: Annotated[AuthenticatedUser, Depends(validate_api_key)]):
    """
    The route receives the identity object directly.
    It knows WHO is calling, not just that auth passed.
    This enables per-user data filtering without a second DB lookup.
    """
    return {
        "caller": auth.user_id,
        "scopes": auth.scopes,
        "orders": [],  # Fetch filtered by auth.user_id in production
    }


@admin_router.get("/metrics")
def get_internal_metrics(auth: Annotated[AuthenticatedUser, Depends(require_admin)]):
    """
    Only admin-role callers reach this handler.
    require_admin handles both auth and role — the route is clean.
    """
    return {"metrics": "[REDACTED]", "caller": auth.user_id}

from fastapi import FastAPI, Depends, HTTPException, status
from sqlalchemy import create_engine, event, text
from sqlalchemy.orm import sessionmaker, Session
from sqlalchemy.pool import QueuePool
from typing import Annotated, Generator
import logging

logger = logging.getLogger(__name__)

SQLALCHEMY_DATABASE_URL = "postgresql://user:password@localhost/forge_db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL,
    poolclass=QueuePool,
    pool_size=20,         # Base connections kept open
    max_overflow=10,      # Additional connections allowed under burst load
    pool_recycle=300,     # Recycle connections after 5 minutes to prevent stale state
    pool_pre_ping=True,   # Test connection health at checkout — catches dropped connections
    pool_timeout=30,      # Raise after 30s waiting for a connection — fail fast, not hang
)

SessionLocal = sessionmaker(
    autocommit=False,
    autoflush=False,
    bind=engine,
)


def get_db() -> Generator[Session, None, None]:
    """
    Yield a database session for the duration of a single request.

    MUST use yield — not return.

    With yield: FastAPI holds the generator open across the request
    lifecycle. After the route completes (or raises), FastAPI resumes
    the generator, triggering the finally block that closes the session.

    Without yield (plain try/finally): FastAPI's exception handling
    intercepts HTTPException before the finally block executes.
    The session is never closed. The connection leaks. The pool
    exhausts. The service returns 503.

    This is the most common FastAPI production bug — and the most
    preventable one.
    """
    db = SessionLocal()
    logger.debug(f"DB session opened: id={id(db)}")
    try:
        # Setup phase: session is open and ready
        yield db
        # If the route commits explicitly, nothing to do here
        # If the route did not commit, nothing is committed — no surprise writes
    except Exception:
        # Roll back any uncommitted transaction before closing
        # Prevents partial writes from surviving an exception
        db.rollback()
        raise
    finally:
        # Teardown phase: ALWAYS runs — success, HTTPException, or crash
        logger.debug(f"DB session closed: id={id(db)}")
        db.close()


# Type alias — use this in route signatures for readability
DBSession = Annotated[Session, Depends(get_db)]


app = FastAPI()


@app.get("/db-status")
def check_db_connection(db: DBSession):
    """
    Health check that validates the DB session is functional.
    Uses db.execute() to confirm the connection is active,
    not just that the session object was created.
    """
    try:
        db.execute(text("SELECT 1"))
        return {"status": "connected", "session_id": id(db)}
    except Exception as e:
        raise HTTPException(
            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
            detail=f"Database unreachable: {str(e)}",
        )


@app.get("/orders/{order_id}")
def get_order(order_id: int, db: DBSession):
    """
    Route that uses the injected session for a real query.
    Whether this raises 404, 422, or succeeds — the session is closed.
    FastAPI guarantees teardown via the yield generator resume.
    """
    # order = db.query(Order).filter(Order.id == order_id).first()
    # if not order:
    #     raise HTTPException(status_code=404, detail="Order not found")
    # return order
    return {"order_id": order_id, "status": "fetched", "session_id": id(db)}


# Example of what NOT to do — shown explicitly so the bug is recognizable in review:
# def get_db_wrong():
#     db = SessionLocal()
#     try:
#         return db  # Returns immediately — FastAPI has no generator to resume
#     except Exception:
#         db.close()
#         raise
#     finally:
#         db.close()  # This finally block DOES NOT RUN when the route raises HTTPException
#                     # FastAPI intercepts the exception before returning here
#                     # Connection leaks. Pool exhausts. 503s at moderate load.

// io.thecodeforge
from fastapi import FastAPI, Depends, HTTPException, status

app = FastAPI()

# Level 1: token extraction
async def extract_token(authorization: str | None = None):
    if not authorization:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED)
    return authorization.split(" ")[-1]

# Level 2: user lookup depends on token
async def get_current_user(token: str = Depends(extract_token)):
    # imagine DB lookup
    if token != "valid-secret":
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)
    return {"username": "ops-lead", "role": "admin"}

# Level 3: admin check depends on user
async def require_admin(user: dict = Depends(get_current_user)):
    if user["role"] != "admin":
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)
    return user

@app.get("/admin/secrets")
async def read_secrets(admin: dict = Depends(require_admin)):
    return {"secret": "the coffee machine needs descaling again"}

// io.thecodeforge
import pytest
from fastapi.testclient import TestClient
from app.main import app
from app.dependencies import get_db_session, get_current_user

# Stub for DB session
def stub_session():
    return "fake-db-session"

# Stub for auth
def stub_admin_user():
    return {"username": "test-admin", "role": "admin"}

@pytest.fixture
def override_deps():
    app.dependency_overrides[get_db_session] = stub_session
    app.dependency_overrides[get_current_user] = stub_admin_user
    yield
    app.dependency_overrides.clear()

@pytest.mark.usefixtures("override_deps")
def test_admin_endpoint_returns_secrets():
    client = TestClient(app)
    response = client.get("/admin/secrets")
    assert response.status_code == 200
    assert "coffee" in response.json()["secret"]