modern-di-grpc 4.1.5

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])
server = grpc.server(
    futures.ThreadPoolExecutor(max_workers=10),
    interceptors=[DIInterceptor(container)],
)
container.validate()  # optional fail-fast; must come after DIInterceptor registers its providers
greeter_pb2_grpc.add_GreeterServicer_to_server(GreeterService(), server)
server.add_insecure_port("[::]:50051")
server.start()
try:
    server.wait_for_termination()
finally:
    server.stop(grace=5).wait()  # let in-flight RPCs finish before APP finalizers run
    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. A container is open from construction, so there is no .open() call before constructing the interceptor, and neither interceptor opens the root itself. Validation is explicit: call .validate() after constructing the interceptor if you want a broken graph to fail at start-up rather than at the first RPC that touches it. gRPC has no server startup or shutdown hook, so closing the root container is up to you: call close_sync() (or await close_async() on grpc.aio) after the server stops, in a finally as above so Ctrl+C still runs the finalizers.

API

Links

• PyPI
• License

Part of modern-python

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

Browse the full list of templates and libraries in modern-python; the org profile has the categorized index.