FastAPI-Template 实践笔记:以依赖注入管理请求生命周期
这个模板仓库在这里:https://github.com/JiayuXu0/FastAPI-Template 。写它的时候,我最纠结的不是目录怎么摆,而是请求生命周期怎么收口。早期做服务时我干过两件蠢事:一是路由里直接拿全局 Session,commit 到处都是;二是异常时忘回滚,偶尔出现“半落地”的脏数据。服务没挂,但数据开始变得不可信,这比挂掉更可怕。
后来我把依赖注入当成“生命周期管理器”来设计,而不是简单的参数注入。下面就是我在模板里反复迭代后比较稳定的一套做法。
请求级资源:把开始和结束写出来 数据库 Session 是典型的请求级资源,我把 commit/rollback 统一收口在依赖层。这样业务层不会出现 session.commit(),错误路径也不会漏掉回滚。
1 2 3 4 5 6 7 8 9 10 def get_session (): session = SessionLocal() try : yield session session.commit() except Exception: session.rollback() raise finally : session.close()
这段代码没有花哨的抽象,但解决了三个现实问题:连接泄漏、异常路径漏回滚、commit 分散带来的不可控。业务逻辑只需要把“意图”表达清楚,事务收口交给依赖层去完成。
应用级资源:放到 lifespan 里统一管理 Redis 连接池、HTTP Client 这类资源生命周期跟应用一致,不应该每个请求都创建。我把它们放到 lifespan,用 app.state 挂载。依赖只负责“取”,不要在依赖里创建大资源。
1 2 3 4 5 6 7 8 9 10 11 12 from contextlib import asynccontextmanagerfrom fastapi import FastAPI@asynccontextmanager async def lifespan (app: FastAPI ): app.state.redis = await create_redis_pool() app.state.http = HttpClient(timeout=5 ) yield await app.state.redis.close() await app.state.http.aclose() app = FastAPI(lifespan=lifespan)
1 2 3 4 from fastapi import Requestdef get_redis (request: Request ): return request.app.state.redis
这样写的好处是:资源创建时机固定、关闭路径清晰,而且依赖只做一件事——提供资源。
依赖装配:路由只组装,不做业务 我习惯把路由层当“装配层”。它不写业务逻辑,只组装依赖,把请求交给服务层处理。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 class UserRepo : def __init__ (self, session ): self ._db = session def get (self, user_id: int ): return self ._db.query(User).filter (User.id == user_id).first() class UserService : def __init__ (self, repo: UserRepo, cache ): self ._repo = repo self ._cache = cache def get_profile (self, user_id: int ): cached = self ._cache.get(f"user:{user_id} " ) if cached: return cached user = self ._repo.get(user_id) if user: self ._cache.set (f"user:{user_id} " , user, ttl=300 ) return user
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 from fastapi import Dependsdef get_user_repo (session = Depends(get_session ) ): return UserRepo(session) def get_user_service ( repo: UserRepo = Depends(get_user_repo ), cache = Depends(get_redis ), ): return UserService(repo, cache) @router.get("/users/{user_id}" ) def get_user ( user_id: int , service: UserService = Depends(get_user_service ), ): return service.get_profile(user_id)
这套结构的价值在于:业务逻辑可测,依赖替换简单。路由只负责装配,服务层专注业务。
事务边界:commit/rollback 只能出现一次 我一直坚持一个硬规则:commit/rollback 只能出现一次。多处 commit 会让事务边界变得模糊,很难排查。
如果业务需要跨多个操作保持一致性,我会用一个轻量的 Unit of Work 做收口:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 class UnitOfWork : def __init__ (self, session ): self .session = session self .users = UserRepo(session) def __enter__ (self ): return self def __exit__ (self, exc_type, exc, tb ): if exc: self .session.rollback() else : self .session.commit() self .session.close() def get_uow (session = Depends(get_session ) ): return UnitOfWork(session) def create_user (payload, uow: UnitOfWork = Depends(get_uow ) ): with uow: uow.users.create(payload)
这样业务层只表达意图,事务边界仍然统一收口。
请求上下文:把追踪信息当成依赖 我不希望业务函数直接拿 Request,那会让测试难做。更干净的方式是把 request_id、user_id 这些信息封装成上下文对象,再注入服务层。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 from dataclasses import dataclassfrom uuid import uuid4from fastapi import Request@dataclass class RequestContext : request_id: str user_id: str | None def get_context (request: Request ) -> RequestContext: request_id = request.headers.get("X-Request-Id" ) or str (uuid4()) user_id = getattr (request.state, "user_id" , None ) return RequestContext(request_id=request_id, user_id=user_id)
这样做的好处是:上下文结构稳定,日志和追踪字段可以在服务层直接使用,单测也不用造 Request。
测试:依赖替换要轻 依赖注入最直接的收益就是测试更轻。我的习惯是:用 dependency_overrides 替换最外层的资源依赖。
1 2 3 4 5 6 7 8 def override_get_session (): session = TestSessionLocal() try : yield session finally : session.close() app.dependency_overrides[get_session] = override_get_session
如果服务层足够纯,测试就简单得像普通函数:
1 2 3 4 5 def test_get_profile (cache, session ): repo = UserRepo(session) service = UserService(repo, cache) result = service.get_profile(1 ) assert result is not None
最后一点取舍 依赖注入很容易被用过头。我的原则是:依赖层只做“资源装配”,不要在依赖里写业务判断;依赖入口集中管理,避免散落。这样项目在半年后不会变成迷宫。
写 FastAPI-Template 的过程里,我越来越确定一件事:依赖注入不是语法糖,它是系统边界的表达方式。边界清楚了,服务才能更稳、更好测,也更容易被团队接手。