停止FastAPI中的属性钻取：使用请求级全局变量

作者头像东城·2026-01-31 23:22:54·
在我的应用程序中，我曾经对自己想出的模式很满意。我知道它有一些缺陷，比如从FastAPI的控制器开始，通过多个层级进行属性钻取。我在路由层创建用户的上下文对象，同样也创建数据库会话等。我基本上是把这些内容一直往下传递，直到领域层的策略类，只是为了回答"我是否被允许做某事"这个问题。
所以我的代码看起来是这样的...
改进前：属性钻取噩梦 🙄
💡
注意context是如何通过多个层传递的，只是为了在最终的策略检查中使用。这是典型的属性钻取 - 通过不需要这些属性的组件传递属性，只是为了让深层嵌套的组件能够访问它们。
  1. # controller.py
  2. @router.post("/projects")
  3. async def create_project(
  4. data: ProjectRequest,
  5. db: Session = Depends(get_db),
  6. context: Context = Depends(get_context)
  7. ):
  8. return await project_service.create_project(data, context)

  10. # services/project_service.py
  11. async def create_project(data: ProjectRequest, context: Context):
  12. # 执行一些业务逻辑
  13. await can_create_project(context).unwrap() # <-- 如果禁止则抛出异常
  14. return await project_repository.create(data, context)

  16. async def can_create_project(context: Context) -> Result[bool, BaseError]:
  17. if not await project_policy.check_create_permission(context):
  18. return Result.err(
  19. error=BadRequestError(
  20. type=ErrorType.PROJECT_CREATION_FORBIDDEN,
  21. description='用户缺乏创建项目的权限',
  22. )
  23. )
  24. return Result.ok(True)

  26. # repositories/project_repository.py
  27. async def create(data: ProjectRequest, context: Context):
  28. # 仍在拖着context到处传递...
  29. project = Project(**data.dict())
  30. context.db.add(project)
  31. # 更多的钻取！
  32. await audit_log.log_creation(project, context)
  33. await event_bus.publish(ProjectCreated(project, context))
  34. return project

  36. # policies/project_policy.py
  37. class ProjectPolicy:
  38. def check_create_permission(self, context: Context):
  39. # 终于！我们在这里实际使用了context
  40. if context.user.role == "admin":
  41. return True
  42. if context.user.organization.plan == "enterprise":
  43. return True
  44. return False
python
灵感来源
然后我偶然在YouTube上看到了DHH关于"Writing Software Well"的播放列表。他提到了全局变量，实际上代码来自Basecamp，是用Ruby编写的。他展示了Ruby中一个非常重要的Current概念。
声明一下：我不太清楚Ruby中的current具体是什么样子，也不知道它是如何工作的。但我试图在我的应用程序中复制相同的功能，这样我就可以删除大量代码，基本上使测试变得更容易。
解决方案
关键在于，在请求的最开始，因为我们是在请求级别而不是整个运行时的全局级别考虑这个概念，只在请求级别。因为单个请求可以基本上分配给单个用户。
所以想法是在请求的最开始创建Python中称为上下文变量的东西。
  1. # context.py
  2. from contextvars import ContextVar
  3. from typing import Optional
  4. from pydantic import BaseModel
  5. from fastapi import Request

  7. class Context(BaseModel):
  8. """
  9. 我们的'当前'上下文 - 类似于Ruby的Current模式。
  10. 包含已认证的用户和常用的引用。
  11. """
  12. issuer_id: str
  13. issuer_role: UserRole
  14. issuer_email: str
  15. issuer_status: UserStatus
  16. organization_id: Optional[str] = None
  17. organization_name: Optional[str] = None
  18. organization_member_role: Optional[OrganizationMemberRole] = None
  19. organization_member_status: Optional[OrganizationMemberStatus] = None
  20. feature_flags: dict[ConfigKey, FeatureFlagConfig] = {}
  21. config: UserConfig
  22. client_info: dict = {}

  24. def is_admin(self) -> bool:
  25. return self.issuer_role == UserRole.ADMIN

  27. def has_feature_flag(self, key: ConfigKey) -> bool:
  28. return self.config.check_ff(key)

  30. def verify_is_admin(self) -> None:
  31. if self.issuer_role != UserRole.ADMIN:
  32. raise ForbiddenError(
  33. type=ErrorType.UNAUTHORIZED,
  34. description='您无权访问此资源',
  35. )

  37. @classmethod
  38. def for_user(cls, user: UserSchema) -> Self:
  39. organization_id = user.organization.id if user.organization else None
  40. organization_member_role = user.organization.role if user.organization else None
  41. return cls(
  42. issuer_id=user.id,
  43. issuer_role=user.role,
  44. issuer_status=user.status,
  45. issuer_email=user.email,
  46. organization_id=organization_id,
  47. organization_name=user.organization.name if user.organization else None,
  48. organization_member_role=organization_member_role,
  49. config=UserConfig(),
  50. )

  52. class _ContextAttributes:
  53. _context: ContextVar[Optional[Context]] = ContextVar('current_context', default=None)

  55. @property
  56. def context(self) -> Context:
  57. ctx = self._context.get()
  58. if ctx is None:
  59. raise RuntimeError('没有可用的上下文 - 不在请求范围内')
  60. return ctx

  62. @property
  63. def user_id(self) -> str:
  64. return self.context.issuer_id

  66. def set(self, context: Context):
  67. self._context.set(context)

  69. def clear(self):
  70. self._context.set(None)

  72. # 全局Current实例
  73. Current = _ContextAttributes()
python
这整个类的实例被设置为Python中的上下文变量。为什么使用上下文变量？如果我们使用异步FastAPI，上下文变量是线程安全的。所以如果同一个实例异步地接收另一个进程，另一个请求，我们不会覆盖这个全局变量。这就是原因。
Python中的上下文变量专门为异步/并发代码设计。每个异步任务都获得自己的上下文副本，防止请求之间的竞争条件和数据泄漏。Current单例模式包装了ContextVar以提供清晰的接口。
现在让我们看看这如何简化我们的代码：
改进后：简洁明了 🎉
  1. # middleware.py
  2. from fastapi import Request
  3. from starlette.middleware.base import BaseHTTPMiddleware

  5. class CurrentContextMiddleware(BaseHTTPMiddleware):
  6. async def dispatch(self, request: Request, call_next):
  7. user = await get_current_user_from_request(request)
  8. context = Context.for_user(user)
  9. context.set_client_info(request)

  11. Current.set(context)

  13. try:
  14. response = await call_next(request)
  15. return response
  16. finally:
  17. Current.clear()

  19. # controller.py
  20. @router.post("/projects")
  21. async def create_project(data: ProjectRequest):
  22. return await project_service.create_project(data)

  24. # services/project_service.py
  25. async def create_project(data: ProjectRequest):
  26. await can_create_project().unwrap() # 如果禁止仍会抛出异常
  27. return await project_repository.create(data)

  29. async def can_create_project() -> Result[bool, BaseError]:
  30. if not await project_policy.check_create_permission():
  31. return Result.err(
  32. error=BadRequestError(
  33. type=ErrorType.PROJECT_CREATION_FORBIDDEN,
  34. description='用户缺乏创建项目的权限',
  35. )
  36. )
  37. return Result.ok(True)

  39. # repositories/project_repository.py
  40. async def create(data: ProjectRequest):
  41. context = Current.context

  43. project = Project(**data.dict())
  44. context.db_session.add(project)

  46. await audit_log.log_creation(project)
  47. await event_bus.publish(ProjectCreated(project))
  48. return project

  50. # policies/project_policy.py
  51. class ProjectPolicy:
  52. def check_create_permission(self):
  53. # 需要时直接访问当前上下文！
  54. context = Current.context

  56. if context.is_admin():
  57. return True
  58. if context.has_feature_flag(ConfigKey.ENTERPRISE_PROJECTS):
  59. return True
  60. if context.organization_member_role == OrganizationMemberRole.OWNER:
  61. return True
  62. return False
python
现在要做的简单事情就是停止传递这个上下文，因为我们已经设置好了。只需要在需要的时候访问它。当它在策略中的最后需要时。当只是检查谁是在请求的人时。这只是代码的10%。所以我们不需要通过所有层把它钻到最底层。
结论
实施这种模式后：
服务和仓储层的代码减少40%
测试设置减少一半 - 不再需要模拟链
就是这样。一个简单的技巧基本上删除了大量代码，它是可维护的，让一切变得更容易，在一些上下文魔法背后隐藏了一些复杂性。
感谢DHH的这个播放列表。我认为整个系列都值得一看。