在我的应用程序中，我曾经对自己想出的模式很满意。我知道它有一些缺陷，列如从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%

• 测试设置减少一半 – 不再需要模拟链

就是这样。一个简单的技巧基本上删除了大量代码，它是可维护的，让一切变得更容易，在一些上下文魔法背后隐藏了一些复杂性。