FastAPI 依赖注入与请求处理机制

v2 重构记录：旧版本使用 http.server.ThreadingHTTPServer（stdlib），BaseHandler → TaskAPIHandler 继承链。v0.2 迁移到 FastAPI + Uvicorn，采用 ASGI 异步架构，依赖注入替代继承。

1. 架构变更概览

方面	v0.1（旧）	v0.2（新）
Web 框架	`http.server.BaseHTTPRequestHandler`	FastAPI ASGI 框架
服务器	`ThreadingHTTPServer`	Uvicorn (uvloop + httptools)
实例化	框架为每个请求 new handler 实例	FastAPI 内部路由分发 + Depends()
路由	正则匹配 `self.path`	装饰器 `@router.get("/{task_id}")`
数据库	JSON 文件 `load_tasks()/save_tasks()`	SQLAlchemy 2.0 async + SQLite
依赖管理	继承覆盖 `handle_api()`	`Depends(get_db)` 依赖注入
API 文档	无	自动 OpenAPI `/docs`
数据验证	无	Pydantic v2 Schema

2. 请求处理时序

FastAPI 请求处理链路：Uvicorn → Starlette → FastAPI Router → 依赖注入 → 业务逻辑

3. FastAPI 依赖注入

3.1 依赖链

FastAPI 使用 Depends() 声明式注入依赖链：

# app/dependencies.py
from typing import Annotated
from fastapi import Depends
from app.database import get_db

DBSession = Annotated[AsyncSession, Depends(get_db)]

# app/routers/tasks.py
@router.get("/{task_id}")
async def get_task(task_id: str, db: DBSession):
    task = await task_service.get_task_by_id(db, task_id)
    ...

请求进来时，FastAPI 解析路由参数 → 调用 get_db() 创建数据库会话 → 注入到 db 参数 → 响应结束后自动关闭会话。

3.2 依赖注入 vs 继承模式对比

维度	旧（继承）	新（DI）
代码位置	`self.handle_api()` 覆盖基类方法	函数签名声明 `db: DBSession`
测试	需要创建完整 handler 实例	替换 `dependency_overrides`即可
可组合性	单继承，复杂时需多级继承	任意组合多个 Depends
类型安全	无类型信息	完整 Pydantic 类型验证
生命周期	请求结束实例销毁	会话自动 close()（context manager）

4. 应用生命周期

FastAPI 使用 lifespan 上下文管理器管理应用启动/关闭：

# app/main.py
@asynccontextmanager
async def lifespan(app: FastAPI):
    await init_db()   # 启动时：创建表
    yield             # 运行中...
    # 关闭时清理资源（如有）

app = FastAPI(title="Task System API", version="0.2.0", lifespan=lifespan)

对比旧版本：旧版本在 server.py:main() 中简单启动 serve_forever()，无生命周期管理。

5. 核心代码

app/main.py — 应用入口

@asynccontextmanager
async def lifespan(app: FastAPI):
    await init_db()
    yield

app = FastAPI(title="Task System API", version="0.2.0", lifespan=lifespan)
app.add_middleware(CORSMiddleware, allow_origins=["*"], ...)
app.include_router(tasks_router)
app.mount("/assets", StaticFiles(directory=FRONTEND_DIST / "assets"))

def main():
    uvicorn.run("app.main:app", host="0.0.0.0", port=8088)

app/database.py — 数据库会话

engine = create_async_engine(settings.database_url, echo=False)
async_session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

async def get_db() -> AsyncSession:
    async with async_session() as session:
        try:
            yield session
        finally:
            await session.close()

app/routers/tasks.py — API 路由

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

@router.get("", response_model=list[TaskResponse])
async def list_tasks(db: DBSession, department: Optional[str] = Query(None)):
    return await task_service.get_tasks(db, department=department)

@router.post("/{task_id}/notes", response_model=TaskResponse)
async def add_note(task_id: str, data: TaskUpdateNote, db: DBSession):
    task = await task_service.add_update_note(db, task_id, data)
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")
    return task

测试覆盖验证：通过 app.dependency_overrides[get_db] = override 可替换数据库会话为 SQLite 内存库，11 个自动化测试全部通过（0.08s）。