grpclib Documentation

Deadlines in grpclib function as propagated timeouts. They allow a timeout constraint to be passed through a chain of services to ensure the entire call chain respects the initial time limit.

The Deadline Lifecycle:

1. A service receives a request with a grpc-timeout in the metadata (e.g., 100m for 100 milliseconds).
2. The service converts this into an absolute deadline: deadline = time.monotonic() + grpc_timeout.
3. When making an outgoing request to another service, the service calculates the remaining time: new_timeout = max(deadline - time.monotonic(), 0).
4. The outgoing request is sent with the new grpc-timeout metadata.

This mechanism allows for simultaneous cancellation of an entire call chain, even in the event of network failures or broken connections.

# Example of converting timeout to deadline and calculating remaining time
import time

# 1. Convert incoming timeout to absolute deadline
grpc_timeout = 0.1  # 100ms
deadline = time.monotonic() + grpc_timeout

# 2. Simulate work
time.sleep(0.02)  # 20ms work

# 3. Calculate remaining timeout for the next service
new_timeout = max(deadline - time.monotonic(), 0)  # Result: ~0.08s (80ms)

The grpclib.events system follows these rules:

1. Property Types:
   • Mutable: Properties you can change (e.g., event.metadata) that will affect the actual gRPC operation.
   • Read-only: Properties that can only be read.
2. Callback Execution: Callbacks are executed in the order they were added (First-In, First-Out).
3. Interrupting Sequences: A callback can call event.interrupt() to stop the sequence of calls for a particular event. This is often used to provide a custom RPC handler (e.g., returning an error instead of the intended method).

from grpclib.events import RecvRequest
from grpclib.exceptions import GRPCError
from grpclib.server import Status

async def authn_error(stream):
    # Custom handler that returns an error
    raise GRPCError(Status.UNAUTHENTICATED)

async def recv_request(event: RecvRequest):
    if event.metadata.get('auth-token') != SECRET:
        # Replace the method function with the error handler
        event.method_func = authn_error
        # Stop the normal execution flow
        event.interrupt()

listen(server, RecvRequest, recv_request)

Beyond standard status codes and messages, you can send rich error details using the google.rpc.Status message format. This allows you to attach structured metadata (like field violations) to an error.

Setup

To enable automatic decoding of these rich details, you must install the following package:

$ pip3 install googleapis-common-protos

Important: Decoding Details

To automatically decode error details, you must import the specific message types you expect to receive. If you do not import them, the error details will appear as stubs (e.g., Unknown('google.rpc.QuotaFailure')) instead of the actual objects.

Server-Side: Sending Details

When raising a GRPCError on the server, pass a list of messages as the third argument to include rich details.

Client-Side: Handling Details

On the client, catch GRPCError and iterate through the details attribute, using isinstance() to identify specific error types.

### Server-Side Example
from google.rpc.error_details_pb2 import BadRequest

async def Method(self, stream):
    ...
    raise GRPCError(
        Status.INVALID_ARGUMENT,
        'Request validation failed',
        [
            BadRequest(
                field_violations=[
                    BadRequest.FieldViolation(
                        field='title',
                        description='This field is required',
                    ),
                ],
            ),
        ],
    )

### Client-Side Example
from google.rpc.error_details_pb2 import BadRequest

try:
    reply = await stub.Method(Request(...))
except GRPCError as err:
    if err.details:
        for detail in err.details:
            if isinstance(detail, BadRequest):
                for violation in detail.field_violations:
                    print(f'{violation.field}: {violation.description}')

You can provide granular health status for specific gRPC services by mapping service instances (or names) to lists of check callables in the Health constructor.

• A specific service is SERVING if all its associated checks pass.
• The OVERALL status is determined by the checks assigned to the OVERALL key. If OVERALL is not explicitly defined, its status depends on all checks registered across all services.
• You can override the OVERALL check list to include a specific subset of checks that do not represent every single service check.

foo = FooService()
bar = BarService()

# Mapping specific services to specific checks
health = Health({
    foo: [a_check, b_check],
    bar: [b_check, c_check],
    OVERALL: [a_check, c_check], # Overriding overall status logic
})

# Test specific service health
$ grpc_health_probe -addr=localhost:50051 -service acme.FooService

# Test overall health
$ grpc_health_probe -addr=localhost:50051