在Strawberry自定义字段解析器中获取层级上下文的问题

问题分析 #

你需要为Resource对象的lastAccessed字段编写自定义解析器，但该字段的值依赖于所属Consumer的ID——同一个Resource被不同Consumer关联时，对应的last_accessed值不同，而当前解析器仅能拿到Resource本身，无法直接获取所属的ConsumerID。

解决方案 #

方法一：在父级解析器中传递关联ID

修改Consumer的resources字段解析器，将当前Consumer的ID附加到返回的每个Resource对象上，这样lastAccessed解析器就能直接获取该值：

def resolve_resources(parent: Any, info: Info):
    # 获取当前Consumer的ID
    consumer_id = parent.id
    # 查询该Consumer关联的所有Resource
    resources = fetch_resources_by_consumer(consumer_id)
    
    # 给每个Resource对象附加Consumer ID属性（用内部命名避免冲突）
    for resource in resources:
        setattr(resource, "_associated_consumer_id", consumer_id)
    
    return resources

然后在lastAccessed的解析器中读取该属性：

def resolve_last_accessed(parent: Any, info: Info):
    resource_id = getattr(parent, "id", None)
    consumer_id = getattr(parent, "_associated_consumer_id", None)
    
    if not resource_id or not consumer_id:
        return None
    
    # 根据双ID查询数据库获取last_accessed
    return fetch_last_accessed(consumer_id, resource_id)

这种方法简单直接，无需改动Schema，适合大多数场景。

方法二：通过上下文维护父对象栈

如果无法修改父级解析器，可利用GraphQL的context对象维护一个父对象栈，自动记录解析路径中的父对象：

首先实现一个中间件来管理栈的压入和弹出：

class ParentStackMiddleware:
    def resolve(self, next_resolver, root, info, **kwargs):
        # 将当前父对象压入栈
        info.context.setdefault("parent_stack", []).append(root)
        try:
            # 执行后续解析
            result = next_resolver(root, info, **kwargs)
            # 处理异步解析场景
            if isinstance(result, AsyncIterable):
                async def wrapped_iterable():
                    async for item in result:
                        yield item
                return wrapped_iterable()
            return result
        finally:
            # 解析完成后弹出栈顶
            info.context["parent_stack"].pop()

将该中间件注册到你的GraphQL服务中后，在lastAccessed解析器里从栈中找到对应的Consumer：

def resolve_last_accessed(parent: Any, info: Info):
    parent_stack = info.context.get("parent_stack", [])
    # 从栈底向上查找最近的Consumer对象
    consumer = next(
        (item for item in reversed(parent_stack) 
         if hasattr(item, "id") and getattr(item, "__typename", "") == "Consumer"),
        None
    )
    
    if not consumer or not parent.id:
        return None
    
    return fetch_last_accessed(consumer.id, parent.id)

方法三：修改Schema引入中间关联对象

如果允许调整Schema结构，推荐新增一个中间类型（如ConsumerResource），直接包含Resource和对应的lastAccessed字段，避免跨层级依赖：

type Consumer {
  id: ID!
  resourceAssociations: [ConsumerResource!]! # 替换原来的resources字段
}

type ConsumerResource {
  resource: Resource!
  lastAccessed: String!
}

type Resource {
  id: ID!
}

然后在Consumer的resourceAssociations解析器中直接查询中间表：

def resolve_resource_associations(parent: Any, info: Info):
    consumer_id = parent.id
    # 查询Consumer-Resource关联表，获取包含last_accessed的记录
    associations = db.query(ConsumerResourceLink).filter_by(consumer_id=consumer_id).all()
    
    # 将数据库记录转换为ConsumerResource类型的对象
    return [
        {"resource": assoc.resource, "lastAccessed": assoc.last_accessed}
        for assoc in associations
    ]

这种方式逻辑最清晰，完全避免了跨解析器传递上下文的问题，适合长期维护。

内容的提问来源于stack exchange，提问作者Nimrod Dolev