gRPC Interceptor

0.15.4 · active · verified Sat Apr 04

grpc-interceptor is a Python library that simplifies the implementation of gRPC interceptors. It provides base classes and utility interceptors that offer direct access to request, response, and service context objects, which are typically harder to access with standard `grpc` library interceptors. The library emphasizes a small, readable codebase and minimal dependencies, primarily `grpcio`. It is actively maintained with regular minor releases, currently at version 0.15.4.

Warnings

Install

Imports

Quickstart

This quickstart demonstrates how to create a custom server interceptor using `grpc-interceptor` to handle exceptions and set gRPC status codes. It includes a basic `ServerInterceptor` subclass that catches `GrpcException` (like `NotFound`) and maps them to gRPC status codes. The example simulates a gRPC call flow to show both successful and error handling paths.

import grpc
from concurrent import futures
from grpc_interceptor import ServerInterceptor
from grpc_interceptor.exceptions import GrpcException, NotFound

# Assuming a generated protobuf service like my_pb2_grpc and my_pb2
# For demonstration, we'll mock these:
class MockRequest:
    def __init__(self, name):
        self.name = name
class MockResponse:
    def __init__(self, message):
        self.message = message
class MockServicerContext:
    def __init__(self):
        self.code = grpc.StatusCode.OK
        self.details = ''
    def set_code(self, code):
        self.code = code
    def set_details(self, details):
        self.details = details
    def abort(self, code, details):
        self.set_code(code)
        self.set_details(details)
        raise GrpcException(code, details)

class CustomExceptionInterceptor(ServerInterceptor):
    def intercept(self, method, request, context, method_name):
        try:
            return method(request, context)
        except GrpcException as e:
            context.set_code(e.status_code)
            context.set_details(e.details)
            raise # Re-raise to let gRPC handle it after context is set
        except Exception as e:
            # Catch other unexpected exceptions
            context.set_code(grpc.StatusCode.INTERNAL)
            context.set_details(f"An unexpected error occurred: {e}")
            raise

class MyServiceServicer:
    def SayHello(self, request, context):
        if request.name == "Error":
            raise NotFound("Name not found!")
        return MockResponse(message=f"Hello, {request.name}!")

def serve():
    # In a real application, you'd use generated stubs and an actual gRPC server
    interceptors = [CustomExceptionInterceptor()]
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10), interceptors=interceptors)
    
    # In a real app, you'd add your service here:
    # my_pb2_grpc.add_MyServiceServicer_to_server(MyServiceServicer(), server)
    
    # For this example, we'll simulate the service call through the interceptor
    service_instance = MyServiceServicer()
    print("Simulating RPC calls through the interceptor:")

    # Successful call
    req_success = MockRequest("World")
    ctx_success = MockServicerContext()
    try:
        res_success = interceptors[0].intercept(service_instance.SayHello, req_success, ctx_success, "/MyService/SayHello")
        print(f"Success: {res_success.message} (Status: {ctx_success.code.name})")
    except Exception as e:
        print(f"Unexpected error in successful call: {e}")

    # Error call
    req_error = MockRequest("Error")
    ctx_error = MockServicerContext()
    try:
        interceptors[0].intercept(service_instance.SayHello, req_error, ctx_error, "/MyService/SayHello")
    except GrpcException as e:
        print(f"Caught expected error: {ctx_error.details} (Status: {ctx_error.code.name})")
    except Exception as e:
        print(f"Caught unexpected error type: {e}")

if __name__ == '__main__':
    serve()

view raw JSON →