Dependency Injection

def Depends(dependency: Optional[Callable[..., Any]] = None, *, use_cache: bool = True) -> Any:
    """
    Declare a dependency for automatic injection and resolution.

    Parameters:
    - dependency: Callable to be invoked as a dependency (function, class, or callable)
                 If None, uses the type annotation of the parameter as the dependency
    - use_cache: Whether to cache the result of this dependency per request
                Can be set to False to call the dependency function multiple times per request

    Returns:
    Dependency marker that FastAPI uses for automatic resolution

    Behaviors:
    - Dependencies are resolved recursively (dependencies can have their own dependencies)  
    - Results are cached per request by default to avoid multiple calls
    - Dependencies are resolved before calling the endpoint function
    - Failed dependencies prevent endpoint execution and return appropriate HTTP errors
    - Sub-dependencies are resolved in dependency order
    """

def Security(
    dependency: Optional[Callable[..., Any]] = None,
    *,
    scopes: Optional[Sequence[str]] = None,
    use_cache: bool = True
) -> Any:
    """
    Declare a security dependency with optional OAuth2 scopes.

    Parameters:
    - dependency: Security callable (typically an OAuth2 or API key scheme)
    - scopes: List of OAuth2 scopes required for this endpoint
    - use_cache: Whether to cache the result of this dependency per request

    Returns:
    Security dependency marker for automatic resolution

    Behaviors:
    - Integrates with OpenAPI security schemas
    - Supports OAuth2 scopes for fine-grained permission control
    - Automatically generates OpenAPI security requirements
    - Can be combined with regular Depends() for complex security patterns
    """

def get_database_connection():
    """
    Example function dependency that provides database connection.
    
    Returns:
    Database connection object
    """

def get_current_user(token: str = Depends(get_auth_token)):
    """
    Example function dependency with sub-dependency.
    
    Parameters:
    - token: Authentication token from sub-dependency
    
    Returns:
    Current user object
    """

class DatabaseManager:
    def __init__(self, connection_string: str = Depends(get_connection_string)):
        """
        Example class dependency with constructor injection.
        
        Parameters:
        - connection_string: Database connection string from dependency
        """
        
    def get_connection(self):
        """Get database connection."""

async def get_async_resource():
    """
    Example async dependency for I/O operations.
    
    Returns:
    Async resource or data
    """

async def validate_async_token(token: str = Depends(get_token)):
    """
    Example async dependency with token validation.
    
    Parameters:
    - token: Token to validate
    
    Returns:
    Validated user information
    
    Raises:
    HTTPException: If token is invalid
    """

def get_database_session():
    """
    Example generator dependency with automatic cleanup.
    
    Yields:
    Database session object
    
    Note: Cleanup code runs after endpoint execution
    """

async def get_async_session():
    """
    Example async generator dependency with automatic cleanup.
    
    Yields:
    Async database session object
    """

from fastapi import FastAPI, Depends, HTTPException

app = FastAPI()

# Simple dependency function
def get_database():
    # In real app, this would return actual database connection
    return {"type": "database", "status": "connected"}

def get_current_user(db = Depends(get_database)):
    # Simulate user lookup
    user = db.get("current_user")  # This would be actual DB query
    if not user:
        raise HTTPException(status_code=401, detail="User not found")
    return {"user_id": 1, "username": "john", "is_active": True}

@app.get("/users/me")
async def get_user_profile(current_user: dict = Depends(get_current_user)):
    return current_user

@app.get("/items/")
async def get_items(
    db = Depends(get_database),
    current_user: dict = Depends(get_current_user)
):
    # Both dependencies are resolved automatically
    return {
        "items": ["item1", "item2"],
        "user": current_user["username"],
        "db_status": db["status"]
    }

from fastapi import FastAPI, Depends

app = FastAPI()

class Settings:
    def __init__(self):
        self.api_key = "secret-key"
        self.debug = True
        self.max_connections = 100

class DatabaseManager:
    def __init__(self, settings: Settings = Depends(Settings)):
        self.settings = settings
        self.connection_pool = f"pool-{settings.max_connections}"
    
    def get_connection(self):
        return f"connection-{self.settings.api_key}"

@app.get("/config")
async def get_config(
    settings: Settings = Depends(Settings),
    db: DatabaseManager = Depends(DatabaseManager)
):
    return {
        "debug": settings.debug,
        "connection": db.get_connection()
    }

from fastapi import FastAPI, Depends, HTTPException, Security
from fastapi.security import HTTPBearer, OAuth2PasswordBearer

app = FastAPI()

# HTTP Bearer token security
security = HTTPBearer()

def verify_token(token: str = Depends(security)):
    if token.credentials != "valid-token":
        raise HTTPException(status_code=401, detail="Invalid token")
    return token.credentials

# OAuth2 with scopes
oauth2_scheme = OAuth2PasswordBearer(
    tokenUrl="token",
    scopes={"read": "Read access", "write": "Write access"}
)

def get_current_user(token: str = Depends(oauth2_scheme)):
    # Verify token and return user
    return {"user_id": 1, "username": "john", "scopes": ["read", "write"]}

def require_scopes(required_scopes: list):
    def check_scopes(current_user: dict = Security(get_current_user, scopes=required_scopes)):
        user_scopes = current_user.get("scopes", [])
        for scope in required_scopes:
            if scope not in user_scopes:
                raise HTTPException(status_code=403, detail="Insufficient permissions")
        return current_user
    return check_scopes

@app.get("/public")
async def public_endpoint():
    return {"message": "This is public"}

@app.get("/protected")
async def protected_endpoint(user: dict = Depends(get_current_user)):
    return {"message": f"Hello {user['username']}"}

@app.post("/admin/users")
async def create_user(user: dict = Security(get_current_user, scopes=["write"])):
    return {"message": "User created", "by": user["username"]}

import asyncio
from fastapi import FastAPI, Depends, HTTPException

app = FastAPI()

async def get_async_database():
    # Simulate async database connection
    await asyncio.sleep(0.1)
    return {"type": "async_db", "status": "connected"}

async def get_external_data():
    # Simulate external API call
    await asyncio.sleep(0.2)
    return {"data": "external_value", "timestamp": "2023-01-01"}

async def validate_user_async(
    db = Depends(get_async_database),
    external_data = Depends(get_external_data)
):
    # Perform async validation
    if external_data["data"] != "external_value":
        raise HTTPException(status_code=400, detail="Validation failed")
    
    return {
        "user_id": 1,
        "username": "async_user",
        "validated_at": external_data["timestamp"]
    }

@app.get("/async-endpoint")
async def async_endpoint(
    user = Depends(validate_user_async),
    db = Depends(get_async_database)
):
    return {
        "user": user,
        "db_status": db["status"]
    }

from fastapi import FastAPI, Depends
import contextlib

app = FastAPI()

def get_database_session():
    """Generator dependency with automatic cleanup."""
    print("Creating database session")
    session = {"id": "session_123", "active": True}
    try:
        yield session
    finally:
        print("Closing database session")
        session["active"] = False

async def get_async_resource():
    """Async generator dependency with cleanup."""
    print("Acquiring async resource")
    resource = {"handle": "resource_456", "open": True}
    try:
        yield resource
    finally:
        print("Releasing async resource")
        resource["open"] = False

@app.get("/with-cleanup")
async def endpoint_with_cleanup(
    db_session = Depends(get_database_session),
    async_resource = Depends(get_async_resource)
):
    # Resources are automatically cleaned up after response
    return {
        "session_id": db_session["id"],
        "resource_handle": async_resource["handle"]
    }

from fastapi import FastAPI, Depends
from fastapi.testclient import TestClient

app = FastAPI()

def get_database():
    return {"type": "production", "host": "prod.db"}

def get_current_user(db = Depends(get_database)):
    return {"user_id": 1, "username": "prod_user"}

@app.get("/data")
async def get_data(user = Depends(get_current_user)):
    return {"data": "sensitive_data", "user": user["username"]}

# Test with dependency overrides
def get_test_database():
    return {"type": "test", "host": "test.db"}

def get_test_user(db = Depends(get_test_database)):
    return {"user_id": 999, "username": "test_user"}

# In tests:
# app.dependency_overrides[get_database] = get_test_database
# app.dependency_overrides[get_current_user] = get_test_user

from fastapi import FastAPI, Depends
import time

app = FastAPI()

def expensive_computation():
    """Expensive operation that should be cached."""
    print("Performing expensive computation...")
    time.sleep(1)  # Simulate expensive operation
    return {"result": "computed_value", "timestamp": time.time()}

def get_user_data(computation = Depends(expensive_computation)):
    return {
        "user_id": 1,
        "computed_result": computation["result"]
    }

def get_admin_data(computation = Depends(expensive_computation)):
    return {
        "admin_id": 1,
        "computed_result": computation["result"]
    }

@app.get("/user-and-admin")
async def get_both_data(
    user_data = Depends(get_user_data),
    admin_data = Depends(get_admin_data)
):
    # expensive_computation() is called only once due to caching
    return {
        "user": user_data,
        "admin": admin_data
    }

# Disable caching for specific dependency
def no_cache_computation():
    print("Computation without caching...")
    return {"result": "fresh_value", "timestamp": time.time()}

def get_fresh_data(computation = Depends(no_cache_computation, use_cache=False)):
    return {"fresh_result": computation["result"]}

@app.get("/fresh-data")
async def get_multiple_fresh(
    data1 = Depends(get_fresh_data),
    data2 = Depends(get_fresh_data)
):
    # no_cache_computation() is called twice due to use_cache=False
    return {"data1": data1, "data2": data2}