Install & Compatibility
Where this runs
tested against v0.15.4 · pip install
no network on importno background threads
Install × environment matrix
Each cell = how many times install + import succeeded across repeated harness runs. Partial = flaky.
glibc = Debian/Ubuntu slim · musl = Alpine Linux
muslpy 3.10–3.910 runs
installs and imports cleanly · install 0.0s · import 0.284s · 40.3MB
glibcpy 3.10–3.910 runs
installs and imports cleanly · install 2.9s · import 0.259s · 39MB
37MB installed
● package 37MB
Code
Verified usage
Verified import paths — ran on the pinned version, not inferred.
ServerInterceptor
✓ from grpc_interceptor import ServerInterceptor
✗ from grpc import ServerInterceptor
Do not confuse with the native `grpc.ServerInterceptor` class, which has a different API.
AsyncServerInterceptor
✓ from grpc_interceptor import AsyncServerInterceptor
ClientInterceptor
✓ from grpc_interceptor import ClientInterceptor
✗ from grpc import ClientInterceptor
Do not confuse with the native `grpc.ClientInterceptor` class, which has a different API.
ExceptionToStatusInterceptor
✓ from grpc_interceptor import ExceptionToStatusInterceptor
AsyncExceptionToStatusInterceptor
✓ from grpc_interceptor import AsyncExceptionToStatusInterceptor
GrpcException
✓ from grpc_interceptor.exceptions import GrpcException
NotFound
✓ from grpc_interceptor.exceptions import NotFound
A specific `GrpcException` subclass for NOT_FOUND status.
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()
Errors
Common errors & fixes
ModuleNotFoundError: No module named 'grpc_interceptor'
The 'grpc-interceptor' library is not installed in the Python environment where the code is being executed.
fixRun `pip install grpc-interceptor` in your terminal to install the library.
AttributeError: 'ServicerContext' object has no attribute 'my_custom_field'
This error occurs when attempting to assign custom attributes directly to the `grpc.ServicerContext` object within an interceptor, which is not supported by the underlying `grpcio` library. The `grpc-interceptor` library provides its own `ServiceContext` wrapper that allows this behavior for `ServerInterceptor` instances.
fixEnsure your interceptor class inherits from `grpc_interceptor.ServerInterceptor` (or `AsyncServerInterceptor`). The `context` object passed to the `intercept` method of these base classes is a `grpc_interceptor.ServiceContext` instance, which supports custom attribute assignment.
```python
from grpc_interceptor import ServerInterceptor
import grpc
class MyCustomDataInterceptor(ServerInterceptor):
def intercept(
self,
method: Callable,
request_or_iterator: Any,
context: grpc.ServicerContext,
method_name: str,
) -> Any:
context.my_custom_field = 'some_value' # This works with grpc-interceptor's ServiceContext
return method(request_or_iterator, context)
``` TypeError: intercept() missing 1 required positional argument: 'method_name'
The `intercept` method in a custom interceptor inheriting from `grpc_interceptor.ServerInterceptor` (or `AsyncServerInterceptor`) has an incorrect or incomplete signature. The base class expects specific arguments for its `intercept` method.
fixAdjust the signature of your `intercept` method to match the required arguments: `self`, `method`, `request_or_iterator`, `context`, and `method_name`.
```python
from grpc_interceptor import ServerInterceptor
from typing import Callable, Any
import grpc
class MyCorrectInterceptor(ServerInterceptor):
def intercept(
self,
method: Callable, # The next interceptor or RPC method implementation
request_or_iterator: Any, # The RPC request or iterator
context: grpc.ServicerContext, # The ServicerContext
method_name: str, # A string like '/package.Service/Method'
) -> Any:
# Your interceptor logic here
return method(request_or_iterator, context)
``` ImportError: cannot import name 'ServerInterceptor' from 'grpc_interceptor'
This error typically indicates that the `ServerInterceptor` class (or other core components like `ClientInterceptor`, `AsyncServerInterceptor`) is being imported incorrectly, either from a non-existent submodule or due to a naming conflict within the project. It could also point to a corrupted installation or an outdated version where the class structure has changed.
fixEnsure that you are importing the interceptor base classes directly from the top-level `grpc_interceptor` package. If the error persists, try reinstalling the library to resolve any potential installation issues.
```python
# Correct imports for server-side interceptors
from grpc_interceptor import ServerInterceptor
from grpc_interceptor import AsyncServerInterceptor
# Correct import for client-side interceptors
from grpc_interceptor import ClientInterceptor
```
Upgrade
Version history
0.15.4latest on PyPI · released Nov 16, 2023
Audit
Dependencies
grpciorequiredCore dependency for gRPC functionality. Version >=1.32.0 is required for async features.
protobufoptionalIncluded with the testing framework extra.