grpc-interceptor

gRPC’s built-in interceptor API in Python is flexible but awkward — it hands interceptors a raw continuation callable and call details rather than direct access to the request, response, or service context, forcing every interceptor to be written against low-level primitives. grpc-interceptor wraps that API in a small set of abstract base classes (ServerInterceptor, AsyncServerInterceptor, ClientInterceptor) so you implement one intercept() method with a natural signature that covers unary and streaming RPCs uniformly, for both sync and asyncio servers.

It also ships ExceptionToStatusInterceptor (and an async variant), which lets service handlers raise typed exceptions like NotFound or PermissionDenied from anywhere in the call stack; the interceptor catches them and sets the correct gRPC status code and details automatically instead of threading context.abort() calls through business logic. A testing submodule provides ready-made dummy gRPC services and client context managers for exercising custom interceptors in unit tests without hand-rolling protobuf fixtures.

What You Get

• Abstract ServerInterceptor / AsyncServerInterceptor base classes with a single intercept() method covering unary and streaming RPCs
• ClientInterceptor base class unifying all four gRPC client interceptor types behind one override point
• ExceptionToStatusInterceptor / AsyncExceptionToStatusInterceptor for mapping raised exceptions to gRPC status codes automatically
• A library of typed exceptions (NotFound, PermissionDenied, AlreadyExists, ResourceExhausted, etc.) mirroring every standard gRPC status code
• A testing submodule with dummy_client/dummy_channel context managers and a pre-built protobuf service for interceptor unit tests
• parse_method_name() and MethodName helpers for extracting the package/service/method from the raw method string gRPC passes to interceptors

Common Use Cases

• Centralizing error handling: raise NotFound/InvalidArgument from deep inside a service handler instead of threading context.abort() calls through every function
• Adding request/response logging or metrics to every RPC without touching individual service implementations
• Injecting auth metadata (bearer tokens, API keys) into outbound gRPC client calls via a ClientInterceptor
• Unit testing custom interceptors against a real gRPC server using the testing module, for both sync and asyncio servers

Under The Hood

Architecture The library is a thin adapter layer over grpc.ServerInterceptor/grpc.aio.ServerInterceptor and the four *ClientInterceptor protocols in grpc_interceptor/server.py and grpc_interceptor/client.py. ServerInterceptor.intercept_service() implements gRPC’s actual continuation-based contract once, unwraps the RPC handler via _get_factory_and_method(), and forwards to a single user-overridden intercept(method, request_or_iterator, context, method_name) method, so subclasses never touch the low-level handler-factory machinery. AsyncServerInterceptor re-implements the same shape for grpc.aio, with extra branching in intercept_service() to distinguish coroutine responses from async-generator streaming responses. ClientInterceptor collapses all four gRPC client interceptor types into one intercept() override via an internal _swap_args() helper. ExceptionToStatusInterceptor composes on top of ServerInterceptor, wrapping the inner call in a context manager (_handle_exception) that catches GrpcException subclasses and calls context.abort(); the module boundary between base interceptor classes and this exception-mapping layer keeps the abstraction small and composable.

Tech Stack Pure Python (3.7+), built on grpcio (^1.49.1) with an optional protobuf (>=4.21.9) extra gated behind the testing extra. Packaging uses Poetry with poetry-core as the build backend. Dev tooling is extensive: nox orchestrates test sessions across Python 3.7–3.11 and a minimum-dependencies check, pytest/pytest-asyncio/pytest-cov run the suite, flake8 (with flake8-bandit, flake8-bugbear, flake8-docstrings, flake8-import-order) and mypy enforce style and typing, and sphinx/xdoctest build documentation with doctested examples. CI (GitHub Actions) runs the matrix across Ubuntu/macOS/Windows and reports coverage to Codecov.

Code Quality The test suite (tests/test_server.py, test_client.py, test_streaming.py, test_exception_to_status.py) exercises both sync and async interceptor paths, including custom-exception handling and multiple streaming RPC shapes, using the library’s own testing module rather than mocks. mypy.ini enables package-wide type checking (with py.typed shipped for downstream consumers), and .flake8 turns on a broad rule set including security linting (bandit) and docstring conventions (Google style), enforced with max-complexity = 10. Public methods carry Google-style docstrings with Args/Returns/Raises sections used to generate Sphinx docs. No gaps were found in test coverage of the public API surface.

API Design The core ergonomic win is collapsing gRPC’s handler-factory/continuation contract into a single method signature per interceptor type, and pairing it with a small hierarchy of named exceptions (one per StatusCode) so status-code selection reads as ordinary exception handling rather than manual context.abort() calls. This is a well-executed ergonomic wrapper rather than a structurally novel approach — the underlying gRPC interceptor model is unchanged — but the exception-to-status mapping and bundled test harness meaningfully reduce boilerplate for anyone writing custom gRPC interceptors in Python.