FastAPI依赖注入：参数共享与逻辑复用

from fastapi import Depends


def auth_service():
    return OAuth2Scheme()


def db_conn(auth: dict = Depends(auth_service)):
    return Database(creds=auth)


@app.get("/data")
async def get_data(conn=Depends(db_conn)):
    return conn.query()

from fastapi import Depends, FastAPI
from sqlalchemy.orm import sessionmaker

SessionLocal = sessionmaker(autocommit=False)


# 请求级作用域
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


# 应用级单例
cache = LRUCache(size=100)


def get_cache():
    return cache

from pydantic import create_model


def dynamic_model(fields: dict):
    return create_model('DynamicModel', **fields)


class FilterFactory:
    @classmethod
    def create(cls, model: BaseModel):
        class QueryParams(model):
            limit: int = 100
            offset: int = 0

        return QueryParams


@app.get("/search")
async def search(params=Depends(FilterFactory.create(User))):
    return params.dict()

from pydantic import validator, root_validator


class GeoValidator:
    @classmethod
    def lat_validator(cls, v):
        if not -90 <= v <= 90:
            raise ValueError("纬度范围错误")
        return v


class Location(BaseModel):
    lat: float
    lng: float

    _validate_lat = validator('lat', allow_reuse=True)(GeoValidator.lat_validator)

class NotificationClient:
    def __init__(self, type: str):
        self.client = self._create_client(type)

    @staticmethod
    def _create_client(type):
        return {
            "sms": SMSClient(),
            "email": EmailClient()
        }[type]


def get_notifier(type: str):
    def _factory():
        return NotificationClient(type)

    return _factory


@app.post("/alert")
async def send_alert(
        notifier: NotificationClient = Depends(get_notifier("sms"))
):
    notifier.client.send()

from fastapi import Header


def feature_flag_dep(feature_name: str):
    class FeatureChecker:
        def __init__(self,
                     enabled: bool = Depends(check_feature_enabled)
                     ):
            if not enabled:
                raise HTTPException(403, "功能未启用")

    return FeatureChecker


def check_feature_enabled(
        feature: str = Header(...),
        config: Config = Depends(get_config)
) -> bool:
    return config.is_enabled(feature)


@app.get("/beta")
async def beta_feature(
        checker=Depends(feature_flag_dep("beta"))
):
    return "功能可用"

class DatabaseError(Exception):
    pass


def db_dep():
    try:
        yield connection
    except Exception as e:
        raise DatabaseError() from e


@app.exception_handler(DatabaseError)
async def handle_db_error(request, exc):
    return JSONResponse(500, {"detail": "数据库异常"})

from fastapi.dependencies.utils import solve_dependencies


def print_dependency_tree():
    routes = app.routes
    for route in routes:
        if isinstance(route, APIRoute):
            solved = solve_dependencies(route.dependant)
            print(f"Route {route.path}:")
            for dep in solved.flat_graph():
                print(f"└─ {dep.call.__name__}")

from fastapi.testclient import TestClient


def override_dep():
    return MockDatabase()


app.dependency_overrides[get_db] = override_dep

client = TestClient(app)
response = client.get("/data")
assert "mock" in response.text

from packaging.version import parse


class VersionedDep:
    def __init__(self, api_version: str = Header(...)):
        self.version = parse(api_version)

    def check_min_version(self, min_version: str):
        if self.version < parse(min_version):
            raise HTTPException(400, "版本过低")


@app.get("/new-feature")
async def new_feature(
        dep: VersionedDep = Depends(),
        checker=Depends(dep.check_min_version("2.3"))
):
    return "功能可用"