Simplified Python gRPC Interceptors¶
The primary aim of this project is to make Python gRPC interceptors simple.
grpc package provides service interceptors, but they’re a bit hard to
use because of their flexibility. The
grpc interceptors don’t have direct access
to the request and response objects, or the service context. Access to these are often
desired, to be able to log data in the request or response, or set status codes on the
The secondary aim of this project is to keep the code small and simple. Code you can
read through and understand quickly gives you confidence and helps debug issues. When
you install this package, you also don’t want a bunch of other packages that might
cause conflicts within your project. Too many dependencies also slow down installation
as well as runtime (fresh imports take time). Hence, a goal of this project is to keep
dependencies to a minimum. The only dependency is the
grpc_interceptor package provides the following:
Interceptorbase class, to make it easy to define your own service interceptors.
ExceptionToStatusInterceptorinterceptor, so your service can raise exceptions that set the gRPC status code correctly (rather than the default of every exception resulting in an
UNKNOWNstatus code). This is something for which pretty much any service will have a use.
$ pip install grpc-interceptor
To define your own interceptor (we can use
ExceptionToStatusInterceptor as an example):
from grpc_interceptor.base import Interceptor class ExceptionToStatusInterceptor(Interceptor): def intercept( self, method: Callable, request: Any, context: grpc.ServicerContext, method_name: str, ) -> Any: """Override this method to implement a custom interceptor. You should call method(request, context) to invoke the next handler (either the RPC method implementation, or the next interceptor in the list). Args: method: The next interceptor, or method implementation. request: The RPC request, as a protobuf message. context: The ServicerContext pass by gRPC to the service. method_name: A string of the form "/protobuf.package.Service/Method" Returns: This should generally return the result of method(request, context), which is typically the RPC method response, as a protobuf message. The interceptor is free to modify this in some way, however. """ try: return method(request, context) except GrpcException as e: context.set_code(e.status_code) context.set_details(e.details) raise
Then inject your interceptor when you create the
interceptors = [ExceptionToStatusInterceptor()] server = grpc.server( futures.ThreadPoolExecutor(max_workers=10), interceptors=interceptors )
from grpc_interceptor.exceptions import NotFound class MyService(my_pb2_grpc.MyServiceServicer): def MyRpcMethod( self, request: MyRequest, context: grpc.ServicerContext ) -> MyResponse: thing = lookup_thing() if not thing: raise NotFound("Sorry, your thing is missing") ...
This results in the gRPC status status code being set to
and the details
"Sorry, your thing is missing". This saves you the hassle of
catching exceptions in your service handler, or passing the context down into
helper functions so they can call
context.set_code. It allows
the more Pythonic approach of just raising an exception from anywhere in the code,
and having it be handled automatically.
These are the current limitations, although supporting these is possible. Contributions or requests are welcome.
Interceptorcurrently only supports unary-unary RPCs.
The package only provides service interceptors.