modern-di-grpc 3.0.2

Modern-DI integration for gRPC (grpcio).

Full guide: gRPC integration docs

Usage example: examples/

Installation

uv add modern-di-grpc      # or: pip install modern-di-grpc

Usage

gRPC has no dependency-injection system of its own, so modern-di-grpc pairs an @inject decorator with inert FromDI markers. A DIInterceptor (sync) or DIAioInterceptor (async) opens one Scope.REQUEST child container per RPC and resolves the FromDI-marked parameters of @inject-decorated servicer methods. Constructing the interceptor registers the ServicerContext provider on the container automatically — there is no separate setup call.

import typing
from concurrent import futures

import grpc
from modern_di import Container, Group, Scope, providers
from modern_di_grpc import DIInterceptor, FromDI, inject

from myapp import greeter_pb2, greeter_pb2_grpc  # your generated stubs


class Settings:
    def __init__(self) -> None:
        self.greeting = "hello"


class Greeter:
    def __init__(self, settings: Settings) -> None:  # auto-injected by type
        self._settings = settings

    def greet(self, name: str) -> str:
        return f"{self._settings.greeting}, {name}"


class AppGroup(Group):
    settings = providers.Factory(Settings, scope=Scope.APP, cache=True)
    greeter = providers.Factory(Greeter, scope=Scope.REQUEST)


class GreeterService(greeter_pb2_grpc.GreeterServicer):
    @inject
    def SayHello(
        self,
        request: greeter_pb2.HelloRequest,
        context: grpc.ServicerContext,
        greeter: typing.Annotated[Greeter, FromDI(Greeter)],  # resolve by type
    ) -> greeter_pb2.HelloReply:
        return greeter_pb2.HelloReply(message=greeter.greet(request.name))


container = Container(groups=[AppGroup])
container.validate()  # optional fail-fast; modern-di 3.1 validates nowhere implicitly
server = grpc.server(
    futures.ThreadPoolExecutor(max_workers=10),
    interceptors=[DIInterceptor(container)],
)
greeter_pb2_grpc.add_GreeterServicer_to_server(GreeterService(), server)
server.add_insecure_port("[::]:50051")
server.start()
server.wait_for_termination()
container.close_sync()

For an async server, pass DIAioInterceptor(container) to grpc.aio.server(...) and write async def servicer methods; @inject adapts to sync, async, and async-generator (server-streaming) methods across all four RPC types. gRPC has no server startup/shutdown hook, so the root container's shutdown is yours to own. As of modern-di 3.1 a container is open from construction, so no .open() call is required before constructing the interceptor (DIInterceptor/DIAioInterceptor never open the root themselves, and no longer need to). Validation is explicit in 3.1: call .validate() if you want a broken graph to fail at start-up rather than at the first RPC that touches it. Call close_sync() (or await close_async() on grpc.aio) after the server stops — that half is still yours, since gRPC gives the adapter no shutdown hook.

API

📦 PyPI

📝 License

Part of modern-python

Built on modern-di, a dependency-injection framework with IoC container and scopes.

Browse the full list of templates and libraries in modern-python — see the org profile for the categorized index.