停止FastAPI中的属性钻取：使用请求级全局变量
在我的应用程序中，我曾经对自己想出的模式很满意。我知道它有一些缺陷，比如从FastAPI的控制器开始，通过多个层级进行属性钻取。我在路由层创建用户的上下文对象，同样也创建数据库会话等。我基本上是把这些内容一直往下传递，直到领域层的策略类，只是为了回答"我是否被允许做某事"这个问题。
所以我的代码看起来是这样的...
改进前：属性钻取噩梦 🙄
💡
注意context是如何通过多个层传递的，只是为了在最终的策略检查中使用。这是典型的属性钻取 - 通过不需要这些属性的组件传递属性，只是为了让深层嵌套的组件能够访问它们。
# controller.py
@router.post("/projects")
async def create_project(
data: ProjectRequest,
db: Session = Depends(get_db),
context: Context = Depends(get_context)
):
return await project_service.create_project(data, context)
# services/project_service.py
async def create_project(data: ProjectRequest, context: Context):
# 执行一些业务逻辑
await can_create_project(context).unwrap() # <-- 如果禁止则抛出异常
return await project_repository.create(data, context)
async def can_create_project(context: Context) -> Result[bool, BaseError]:
if not await project_policy.check_create_permission(context):
return Result.err(
error=BadRequestError(
type=ErrorType.PROJECT_CREATION_FORBIDDEN,
description='用户缺乏创建项目的权限',
)
)
return Result.ok(True)
# repositories/project_repository.py
async def create(data: ProjectRequest, context: Context):
# 仍在拖着context到处传递...
project = Project(**data.dict())
context.db.add(project)
# 更多的钻取！
await audit_log.log_creation(project, context)
await event_bus.publish(ProjectCreated(project, context))
return project
# policies/project_policy.py
class ProjectPolicy:
def check_create_permission(self, context: Context):
# 终于！我们在这里实际使用了context
if context.user.role == "admin":
return True
if context.user.organization.plan == "enterprise":
return True
return False
灵感来源
然后我偶然在YouTube上看到了DHH关于"Writing Software Well"的播放列表。他提到了全局变量，实际上代码来自Basecamp，是用Ruby编写的。他展示了Ruby中一个非常重要的Current概念。
声明一下：我不太清楚Ruby中的current具体是什么样子，也不知道它是如何工作的。但我试图在我的应用程序中复制相同的功能，这样我就可以删除大量代码，基本上使测试变得更容易。
解决方案
关键在于，在请求的最开始，因为我们是在请求级别而不是整个运行时的全局级别考虑这个概念，只在请求级别。因为单个请求可以基本上分配给单个用户。
所以想法是在请求的最开始创建Python中称为上下文变量的东西。
# context.py
from contextvars import ContextVar
from typing import Optional
from pydantic import BaseModel
from fastapi import Request
class Context(BaseModel):
"""
我们的'当前'上下文 - 类似于Ruby的Current模式。
包含已认证的用户和常用的引用。
"""
issuer_id: str
issuer_role: UserRole
issuer_email: str
issuer_status: UserStatus
organization_id: Optional[str] = None
organization_name: Optional[str] = None
organization_member_role: Optional[OrganizationMemberRole] = None
organization_member_status: Optional[OrganizationMemberStatus] = None
feature_flags: dict[ConfigKey, FeatureFlagConfig] = {}
config: UserConfig
client_info: dict = {}
def is_admin(self) -> bool:
return self.issuer_role == UserRole.ADMIN
def has_feature_flag(self, key: ConfigKey) -> bool:
return self.config.check_ff(key)
def verify_is_admin(self) -> None:
if self.issuer_role != UserRole.ADMIN:
raise ForbiddenError(
type=ErrorType.UNAUTHORIZED,
description='您无权访问此资源',
)
@classmethod
def for_user(cls, user: UserSchema) -> Self:
organization_id = user.organization.id if user.organization else None
organization_member_role = user.organization.role if user.organization else None
return cls(
issuer_id=user.id,
issuer_role=user.role,
issuer_status=user.status,
issuer_email=user.email,
organization_id=organization_id,
organization_name=user.organization.name if user.organization else None,
organization_member_role=organization_member_role,
config=UserConfig(),
)
class _ContextAttributes:
_context: ContextVar[Optional[Context]] = ContextVar('current_context', default=None)
@property
def context(self) -> Context:
ctx = self._context.get()
if ctx is None:
raise RuntimeError('没有可用的上下文 - 不在请求范围内')
return ctx
@property
def user_id(self) -> str:
return self.context.issuer_id
def set(self, context: Context):
self._context.set(context)
def clear(self):
self._context.set(None)
# 全局Current实例
Current = _ContextAttributes()
这整个类的实例被设置为Python中的上下文变量。为什么使用上下文变量？如果我们使用异步FastAPI，上下文变量是线程安全的。所以如果同一个实例异步地接收另一个进程，另一个请求，我们不会覆盖这个全局变量。这就是原因。
Python中的上下文变量专门为异步/并发代码设计。每个异步任务都获得自己的上下文副本，防止请求之间的竞争条件和数据泄漏。Current单例模式包装了ContextVar以提供清晰的接口。
现在让我们看看这如何简化我们的代码：
改进后：简洁明了 🎉
# middleware.py
from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware
class CurrentContextMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next):
user = await get_current_user_from_request(request)
context = Context.for_user(user)
context.set_client_info(request)
Current.set(context)
try:
response = await call_next(request)
return response
finally:
Current.clear()
# controller.py
@router.post("/projects")
async def create_project(data: ProjectRequest):
return await project_service.create_project(data)
# services/project_service.py
async def create_project(data: ProjectRequest):
await can_create_project().unwrap() # 如果禁止仍会抛出异常
return await project_repository.create(data)
async def can_create_project() -> Result[bool, BaseError]:
if not await project_policy.check_create_permission():
return Result.err(
error=BadRequestError(
type=ErrorType.PROJECT_CREATION_FORBIDDEN,
description='用户缺乏创建项目的权限',
)
)
return Result.ok(True)
# repositories/project_repository.py
async def create(data: ProjectRequest):
context = Current.context
project = Project(**data.dict())
context.db_session.add(project)
await audit_log.log_creation(project)
await event_bus.publish(ProjectCreated(project))
return project
# policies/project_policy.py
class ProjectPolicy:
def check_create_permission(self):
# 需要时直接访问当前上下文！
context = Current.context
if context.is_admin():
return True
if context.has_feature_flag(ConfigKey.ENTERPRISE_PROJECTS):
return True
if context.organization_member_role == OrganizationMemberRole.OWNER:
return True
return False
现在要做的简单事情就是停止传递这个上下文，因为我们已经设置好了。只需要在需要的时候访问它。当它在策略中的最后需要时。当只是检查谁是在请求的人时。这只是代码的10%。所以我们不需要通过所有层把它钻到最底层。
结论
实施这种模式后：
服务和仓储层的代码减少40%
测试设置减少一半 - 不再需要模拟链
就是这样。一个简单的技巧基本上删除了大量代码，它是可维护的，让一切变得更容易，在一些上下文魔法背后隐藏了一些复杂性。
感谢DHH的这个播放列表。我认为整个系列都值得一看。