linkd.ext.grpc

Extension module adding support for using linkd-based dependency injection with gRPC.

See the examples directory for a full working application using this module.

Note

Unlike the FastAPI and Starlette integrations, the default inject() method will work for method handlers when using this extension.

---

class Contexts

Collection of the dependency injection context values linkd.ext.grpc uses.

REQUEST = 'linkd.contexts.http.request'

DI context used during gRPC request handling.

ROOT = 'linkd.contexts.root'

The root DI context - ALL other contexts are built with this as the parent.

class DiInterceptor(manager: DependencyInjectionManager)

Server interceptor which handles setting up a DI context for each gRPC request.

The ROOT and REQUEST contexts are entered for the duration of every RPC, across all four cardinalities (unary-unary, unary-stream, stream-unary, and stream-stream). Within the request context, the active grpc.ServicerContext is registered as a dependency - as is the request message (google.protobuf.message.Message) for the unary-request RPCs.

Parameters:

manager – The dependency injection manager to use when entering the DI context.

Example

import grpc
import linkd

manager = linkd.DependencyInjectionManager()

server = grpc.aio.server(interceptors=[linkd.ext.grpc.DiInterceptor(manager)])

async intercept_service(continuation: Callable[[grpc.HandlerCallDetails], Awaitable[grpc.RpcMethodHandler[t.Any, t.Any]]], handler_call_details: grpc.HandlerCallDetails) → grpc.RpcMethodHandler[t.Any, t.Any] | None

Intercepts incoming RPCs before handing them over to a handler.

State can be passed from an interceptor to downstream interceptors via contextvars. The first interceptor is called from an empty contextvars.Context, and the same Context is used for downstream interceptors and for the final handler call. Note that there are no guarantees that interceptors and handlers will be called from the same thread.

Parameters:

• continuation – A function that takes a HandlerCallDetails and proceeds to invoke the next interceptor in the chain, if any, or the RPC handler lookup logic, with the call details passed as an argument, and returns an RpcMethodHandler instance if the RPC is considered serviced, or None otherwise.

• handler_call_details – A HandlerCallDetails describing the RPC.

Returns:

An RpcMethodHandler with which the RPC may be serviced if the interceptor chooses to service this RPC, or None otherwise.