Reference¶

grpc_interceptor
grpc_interceptor.exceptions
grpc_interceptor.testing

grpc_interceptor¶

Simplified Python gRPC interceptors.

class grpc_interceptor.ExceptionToStatusInterceptor(status_on_unknown_exception: Optional[grpc.StatusCode] = None)¶

An interceptor that catches exceptions and sets the RPC status and details.

ExceptionToStatusInterceptor will catch any subclass of GrpcException and set the status code and details on the gRPC context.

Parameters: status_on_unknown_exception – Specify what to do if an exception which is not a subclass of GrpcException is raised. If None, do nothing (by default, grpc will set the status to UNKNOWN). If not None, then the status code will be set to this value. It must not be OK. The details will be set to the value of repr(e), where e is the exception. In any case, the exception will be propagated.
Raises: ValueError – If status_code is OK.

intercept(method: Callable, request: Any, context: grpc.ServicerContext, method_name: str) → Any¶: Do not call this directly; use the interceptor kwarg on grpc.server().

class grpc_interceptor.MethodName(package: str, service: str, method: str)¶

Represents a gRPC method name.

gRPC methods are defined by three parts, represented by the three attributes.

package¶: This is defined by the package foo.bar; designation in the protocol buffer definition, or it could be defined by the protocol buffer directory structure, depending on the language (see https://developers.google.com/protocol-buffers/docs/proto3#packages).

service¶: This is the service name in the protocol buffer definition (e.g., service SearchService { … }.

method¶: This is the method name. (e.g., rpc Search(…) returns (…);).

property fully_qualified_service¶

Return the service name prefixed with the package.

Example

>>> MethodName("foo.bar", "SearchService", "Search").fully_qualified_service
'foo.bar.SearchService'

class grpc_interceptor.ServerInterceptor¶

Base class for server-side interceptors.

To implement an interceptor, subclass this class and override the intercept method.

abstract intercept(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).

Parameters

method – Either the RPC method implementation, or the next interceptor in the chain.
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.

intercept_service(continuation, handler_call_details)¶

Implementation of grpc.ServerInterceptor.

This is not part of the grpc_interceptor.ServerInterceptor API, but must have a public name. Do not override it, unless you know what you’re doing.

grpc_interceptor.parse_method_name(method_name: str) → grpc_interceptor.server.MethodName¶

Parse a method name into package, service and endpoint components.

Parameters: method_name – A string of the form “/foo.bar.SearchService/Search”, as passed to ServerInterceptor.intercept().
Returns: A MethodName object.

Example

>>> parse_method_name("/foo.bar.SearchService/Search")
MethodName(package='foo.bar', service='SearchService', method='Search')

grpc_interceptor.exceptions¶

Exceptions for ExceptionToStatusInterceptor.

See https://grpc.github.io/grpc/core/md_doc_statuscodes.html for the source of truth on status code meanings.

exception grpc_interceptor.exceptions.Aborted(details: Optional[str] = None, status_code: Optional[grpc.StatusCode] = None)¶

The operation was aborted.

Typically this is due to a concurrency issue such as a sequencer check failure or transaction abort. See the guidelines on other exceptions for deciding between FAILED_PRECONDITION, ABORTED, and UNAVAILABLE.

exception grpc_interceptor.exceptions.AlreadyExists(details: Optional[str] = None, status_code: Optional[grpc.StatusCode] = None)¶

The entity that a client attempted to create already exists.

E.g., a file or directory that a client is trying to create already exists.

exception grpc_interceptor.exceptions.Cancelled(details: Optional[str] = None, status_code: Optional[grpc.StatusCode] = None)¶: The operation was cancelled, typically by the caller.

exception grpc_interceptor.exceptions.DataLoss(details: Optional[str] = None, status_code: Optional[grpc.StatusCode] = None)¶: Unrecoverable data loss or corruption.

exception grpc_interceptor.exceptions.DeadlineExceeded(details: Optional[str] = None, status_code: Optional[grpc.StatusCode] = None)¶

The deadline expired before the operation could complete.

For operations that change the state of the system, this error may be returned even if the operation has completed successfully. For example, a successful response from a server could have been delayed long.

exception grpc_interceptor.exceptions.FailedPrecondition(details: Optional[str] = None, status_code: Optional[grpc.StatusCode] = None)¶

The operation failed because the system is in an invalid state for execution.

For example, the directory to be deleted is non-empty, an rmdir operation is applied to a non-directory, etc. Service implementors can use the following guidelines to decide between FAILED_PRECONDITION, ABORTED, and UNAVAILABLE: (a) Use UNAVAILABLE if the client can retry just the failing call. (b) Use ABORTED if the client should retry at a higher level (e.g., when a client-specified test-and-set fails, indicating the client should restart a read-modify-write sequence). (c) Use FAILED_PRECONDITION if the client should not retry until the system state has been explicitly fixed. E.g., if an “rmdir” fails because the directory is non-empty, FAILED_PRECONDITION should be returned since the client should not retry unless the files are deleted from the directory.

exception grpc_interceptor.exceptions.GrpcException(details: Optional[str] = None, status_code: Optional[grpc.StatusCode] = None)¶

Base class for gRPC exceptions.

Generally you would not use this class directly, but rather use a subclass representing one of the standard gRPC status codes (see: https://grpc.github.io/grpc/core/md_doc_statuscodes.html for the official list).

status_code¶: A grpc.StatusCode other than OK. The only use case for this is if gRPC adds a new status code that isn’t represented by one of the subclasses of GrpcException. Must not be OK, because gRPC will not raise an RpcError to the client if the status code is OK.

details¶: A string with additional informantion about the error.

Parameters

details – If not None, specifies a custom error message.
status_code – If not None, sets the status code.

Raises

ValueError – If status_code is OK.

property status_string¶

Return status_code as a string.

Returns: The status code as a string.

Example

>>> GrpcException(status_code=StatusCode.NOT_FOUND).status_string
'NOT_FOUND'

exception grpc_interceptor.exceptions.Internal(details: Optional[str] = None, status_code: Optional[grpc.StatusCode] = None)¶

Internal errors.

This means that some invariants expected by the underlying system have been broken. This error code is reserved for serious errors.

exception grpc_interceptor.exceptions.InvalidArgument(details: Optional[str] = None, status_code: Optional[grpc.StatusCode] = None)¶

The client specified an invalid argument.

Note that this differs from FAILED_PRECONDITION. INVALID_ARGUMENT indicates arguments that are problematic regardless of the state of the system (e.g., a malformed file name).

exception grpc_interceptor.exceptions.NotFound(details: Optional[str] = None, status_code: Optional[grpc.StatusCode] = None)¶

Some requested entity (e.g., file or directory) was not found.

Note to server developers: if a request is denied for an entire class of users, such as gradual feature rollout or undocumented whitelist, NOT_FOUND may be used. If a request is denied for some users within a class of users, such as user-based access control, PERMISSION_DENIED must be used.

exception grpc_interceptor.exceptions.OutOfRange(details: Optional[str] = None, status_code: Optional[grpc.StatusCode] = None)¶

The operation was attempted past the valid range.

E.g., seeking or reading past end-of-file. Unlike INVALID_ARGUMENT, this error indicates a problem that may be fixed if the system state changes. For example, a 32-bit file system will generate INVALID_ARGUMENT if asked to read at an offset that is not in the range [0,2^32-1], but it will generate OUT_OF_RANGE if asked to read from an offset past the current file size. There is a fair bit of overlap between FAILED_PRECONDITION and OUT_OF_RANGE. We recommend using OUT_OF_RANGE (the more specific error) when it applies so that callers who are iterating through a space can easily look for an OUT_OF_RANGE error to detect when they are done.

exception grpc_interceptor.exceptions.PermissionDenied(details: Optional[str] = None, status_code: Optional[grpc.StatusCode] = None)¶

The caller does not have permission to execute the specified operation.

PERMISSION_DENIED must not be used for rejections caused by exhausting some resource (use RESOURCE_EXHAUSTED instead for those errors). PERMISSION_DENIED must not be used if the caller can not be identified (use UNAUTHENTICATED instead for those errors). This error code does not imply the request is valid or the requested entity exists or satisfies other pre-conditions.

exception grpc_interceptor.exceptions.ResourceExhausted(details: Optional[str] = None, status_code: Optional[grpc.StatusCode] = None)¶

Some resource has been exhausted.

Perhaps a per-user quota, or perhaps the entire file system is out of space.

exception grpc_interceptor.exceptions.Unauthenticated(details: Optional[str] = None, status_code: Optional[grpc.StatusCode] = None)¶: The request does not have valid authentication credentials for the operation.

exception grpc_interceptor.exceptions.Unavailable(details: Optional[str] = None, status_code: Optional[grpc.StatusCode] = None)¶

The service is currently unavailable.

This is most likely a transient condition, which can be corrected by retrying with a backoff. Note that it is not always safe to retry non-idempotent operations.

exception grpc_interceptor.exceptions.Unimplemented(details: Optional[str] = None, status_code: Optional[grpc.StatusCode] = None)¶: The operation is not implemented or is not supported/enabled in this service.

exception grpc_interceptor.exceptions.Unknown(details: Optional[str] = None, status_code: Optional[grpc.StatusCode] = None)¶

Unknown error.

For example, this error may be returned when a Status value received from another address space belongs to an error space that is not known in this address space. Also errors raised by APIs that do not return enough error information may be converted to this error.

grpc_interceptor.testing¶

A framework for testing interceptors.

class grpc_interceptor.testing.DummyRequest¶

input¶: Field DummyRequest.input

class grpc_interceptor.testing.DummyResponse¶

output¶: Field DummyResponse.output

class grpc_interceptor.testing.DummyService(special_cases: Dict[str, Callable[str, str]])¶

A gRPC service used for testing.

Parameters: special_cases – A dictionary where the keys are strings, and the values are functions that take and return strings. The functions can also raise exceptions. When the Execute method is given a string in the dict, it will call the function with that string instead, and return the result. This allows testing special cases, like raising exceptions.

Execute(request: grpc_interceptor.testing.protos.dummy_pb2.DummyRequest, context: grpc.ServicerContext) → grpc_interceptor.testing.protos.dummy_pb2.DummyResponse¶: Echo the input, or take on of the special cases actions.

ExecuteClientServerStream(request_iter: Iterable[grpc_interceptor.testing.protos.dummy_pb2.DummyRequest], context: grpc.ServicerContext) → Iterable[grpc_interceptor.testing.protos.dummy_pb2.DummyResponse]¶: Stream input to output.

ExecuteClientStream(request_iter: Iterable[grpc_interceptor.testing.protos.dummy_pb2.DummyRequest], context: grpc.ServicerContext) → grpc_interceptor.testing.protos.dummy_pb2.DummyResponse¶: Iterate over the input and concatenates the strings into the output.

ExecuteServerStream(request: grpc_interceptor.testing.protos.dummy_pb2.DummyRequest, context: grpc.ServicerContext) → Iterable[grpc_interceptor.testing.protos.dummy_pb2.DummyResponse]¶: Stream one character at a time from the input.

grpc_interceptor.testing.dummy_client(special_cases: Dict[str, Callable[str, str]], interceptors: List[grpc_interceptor.server.ServerInterceptor])¶: A context manager that returns a gRPC client connected to a DummyService.

grpc_interceptor.testing.raises(e: Exception) → Callable¶

Return a function that raises the given exception when called.

Parameters: e – The exception to be raised.
Returns: A function that can take any arguments, and raises the given exception.