{"library":"grpc-interceptor","code":"import grpc\nfrom concurrent import futures\nfrom grpc_interceptor import ServerInterceptor\nfrom grpc_interceptor.exceptions import GrpcException, NotFound\n\n# Assuming a generated protobuf service like my_pb2_grpc and my_pb2\n# For demonstration, we'll mock these:\nclass MockRequest:\n    def __init__(self, name):\n        self.name = name\nclass MockResponse:\n    def __init__(self, message):\n        self.message = message\nclass MockServicerContext:\n    def __init__(self):\n        self.code = grpc.StatusCode.OK\n        self.details = ''\n    def set_code(self, code):\n        self.code = code\n    def set_details(self, details):\n        self.details = details\n    def abort(self, code, details):\n        self.set_code(code)\n        self.set_details(details)\n        raise GrpcException(code, details)\n\nclass CustomExceptionInterceptor(ServerInterceptor):\n    def intercept(self, method, request, context, method_name):\n        try:\n            return method(request, context)\n        except GrpcException as e:\n            context.set_code(e.status_code)\n            context.set_details(e.details)\n            raise # Re-raise to let gRPC handle it after context is set\n        except Exception as e:\n            # Catch other unexpected exceptions\n            context.set_code(grpc.StatusCode.INTERNAL)\n            context.set_details(f\"An unexpected error occurred: {e}\")\n            raise\n\nclass MyServiceServicer:\n    def SayHello(self, request, context):\n        if request.name == \"Error\":\n            raise NotFound(\"Name not found!\")\n        return MockResponse(message=f\"Hello, {request.name}!\")\n\ndef serve():\n    # In a real application, you'd use generated stubs and an actual gRPC server\n    interceptors = [CustomExceptionInterceptor()]\n    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10), interceptors=interceptors)\n    \n    # In a real app, you'd add your service here:\n    # my_pb2_grpc.add_MyServiceServicer_to_server(MyServiceServicer(), server)\n    \n    # For this example, we'll simulate the service call through the interceptor\n    service_instance = MyServiceServicer()\n    print(\"Simulating RPC calls through the interceptor:\")\n\n    # Successful call\n    req_success = MockRequest(\"World\")\n    ctx_success = MockServicerContext()\n    try:\n        res_success = interceptors[0].intercept(service_instance.SayHello, req_success, ctx_success, \"/MyService/SayHello\")\n        print(f\"Success: {res_success.message} (Status: {ctx_success.code.name})\")\n    except Exception as e:\n        print(f\"Unexpected error in successful call: {e}\")\n\n    # Error call\n    req_error = MockRequest(\"Error\")\n    ctx_error = MockServicerContext()\n    try:\n        interceptors[0].intercept(service_instance.SayHello, req_error, ctx_error, \"/MyService/SayHello\")\n    except GrpcException as e:\n        print(f\"Caught expected error: {ctx_error.details} (Status: {ctx_error.code.name})\")\n    except Exception as e:\n        print(f\"Caught unexpected error type: {e}\")\n\nif __name__ == '__main__':\n    serve()\n","lang":"python","description":"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.","tag":null,"tag_description":null,"last_tested":"2026-04-24","results":[{"runtime":"python:3.10-alpine","exit_code":0},{"runtime":"python:3.10-slim","exit_code":0},{"runtime":"python:3.11-alpine","exit_code":0},{"runtime":"python:3.11-slim","exit_code":0},{"runtime":"python:3.12-alpine","exit_code":0},{"runtime":"python:3.12-slim","exit_code":0},{"runtime":"python:3.13-alpine","exit_code":0},{"runtime":"python:3.13-slim","exit_code":0},{"runtime":"python:3.9-alpine","exit_code":0},{"runtime":"python:3.9-slim","exit_code":0}]}