时间戳

首页 > 解决方案 > gRPC：从装饰器或 ServiceInterceptor 将属性添加到上下文

python - gRPC：从装饰器或 ServiceInterceptor 将属性添加到上下文

问题描述

我正在使用装饰器进行身份验证。如果成功，我想context用更多信息来扩展。如何做到这一点？

@require_authorization
def test(request, context):
    <...>

def require_authorization(func):
    def wrapper(request, context):
        <...How can I extend the context here?...>
        return func(request, context)
    return wrapper

我也尝试过 ServiceInterceptor，但也被卡住了：

class Interceptor(grpc.ServerInterceptor):

    def intercept_service(self, continuation, handler_call_details):
        <...Can I extend the metadata here somehow?...>
        return continuation(handler_call_details)

server = grpc.server(
            futures.ThreadPoolExecutor(max_workers=10),
            interceptors=(Interceptor(),),
            )

使用装饰器的解决方法

根据 Lidi Zheng 的回复（谢谢！）我正在使用以下装饰器。可能最烦人的方面是必须为抽象基类的所有方法提供实现。由于这没有很好的记录，我从源头得到了它们。

class AuthServicerContext(grpc.ServicerContext):

    def __init__(self, context: grpc.ServicerContext, func_name):
        self._original_context = context
        meta = dict(context.invocation_metadata())

    def abort(self, code, details):
        self._original_context.abort(code, details)

    def abort_with_status(self, status):
        self._original_context.abort_with_status(status)

    def add_callback(self, callback):
        self._original_context.add_callback(callback)

    def auth_context(self):
        self._original_context.auth_context()

    def cancel(self):
        self._original_context.cancel()

    def invocation_metadata(self):
        self._original_context.invocation_metadata()

    def is_active(self):
        self._original_context.is_active()

    def peer(self):
        self._original_context.peer()

    def peer_identities(self):
        self._original_context.peer_identities()

    def peer_identity_key(self):
        self._original_context.peer_identity_key()

    def send_initial_metadata(self, initial_metadata):
        self._original_context.send_initial_metadata(initial_metadata)

    def set_code(self,code):
        self._original_context.set_code(code)

    def set_details(self, details):
        self._original_context.set_details(details)

    def set_trailing_metadata(self, trailing_metadata):
        self._original_context.abort(trailing_metadata)

    def time_remaining(self):
        self._original_context.time_remaining()

def require_authorization(func):
    def wrapper(self, request, context):
        return func(self, request, AuthServicerContext(context, func.__name__))
    return wrapper

标签： pythongrpc

解决方案

不幸的是，ServicerContext不支持向直通拦截器添加属性。查看服务器端拦截器的历史。

另一方面，使用装饰器进行身份验证是可能的：

class MyPowerfulServicerContext(grpc.ServicerContext):
    def __init__(self, original_context: grpc.ServicerContext):
        self._original_context = original_context
        ...Perform authentication with metadata / peer identity / etc.

    def ...Your_extended_functions

    # You can port the invocation  of methods to the original context
    def abort(self, code, details):
        self._original_context.abort(code, details)

    ...



def require_authorization(func):
    def wrapper(request, context):
        return func(request, MyPowerfulServicerContext(context))
    return wrapper

希望这对您的用例有所帮助，如果我错过了理解，请发表评论。此外，请随时向https://github.com/grpc/grpc提出问题。

推荐阅读