asyncio gRPC：如何从装饰器/服务拦截器为上下文添加属性？

问题背景 #

在同步gRPC场景下，可通过从invocation_metadata计算信息并传递给服务方法实现逻辑复用，但切换到asyncio gRPC后原有方案失效：

• 直接为context添加属性context.pass_to_func = "computed value"，报错'grpc._cython.cygrpc._ServicerContext'对象无该属性
• 尝试字典式赋值context['pass_to_func'] = "computed value"，提示该对象不支持项赋值

需在异步包装器中为_ServicerContext添加自定义属性后传递给业务方法。

可行解决方案 #

方案1：自定义上下文包装类（推荐） #

通过包装原始_ServicerContext对象，将自定义属性存储在包装类中，同时代理原始context的所有方法与属性，确保业务方法调用无感知：

class ContextWrapper:
    def __init__(self, original_context):
        self._original_context = original_context
        # 初始化自定义属性，可后续赋值计算后的信息
        self.pass_to_func = None

    # 代理原始context的属性访问
    def __getattr__(self, name):
        return getattr(self._original_context, name)

    # 异步方法单独代理（如write_trailing_metadata等）
    async def write_trailing_metadata(self, *args, **kwargs):
        return await self._original_context.write_trailing_metadata(*args, **kwargs)

    # 按需添加其他异步方法的代理

在异步拦截器或装饰器中替换原始context：

async def async_request_wrapper(handler, request, context):
    # 从invocation_metadata计算所需信息
    computed_value = "从metadata解析出的计算结果"
    # 包装原始context
    wrapped_context = ContextWrapper(context)
    wrapped_context.pass_to_func = computed_value
    # 调用业务方法
    return await handler(request, wrapped_context)

方案2：利用自定义metadata临时存储（应急方案） #

若不想引入包装类，可将计算后的信息存入invocation_metadata，通过自定义key区分业务数据：

# 在包装器中添加自定义metadata
computed_value = "计算后的字符串数据"
# 异步context需调用invocation_metadata()获取元数据列表
updated_metadata = list(context.invocation_metadata()) + [("x-custom-transit-data", computed_value)]

# 业务方法中读取自定义数据
def extract_custom_data(context):
    for key, value in context.invocation_metadata():
        if key == "x-custom-transit-data":
            return value
    return None

该方案仅适用于字符串类型数据，且易与业务元数据混淆，不推荐用于复杂场景。

方案3：动态扩展Cython类（不推荐） #

由于_ServicerContext是Cython实现的内置类，无法直接动态添加属性，此类尝试（如使用setattr或第三方库）大概率会触发兼容性问题，不建议采用。

内容的提问来源于stack exchange，提问作者Tejo Vardhan Makineni