modern-di-grpc 3.0.0

Modern-DI integration for gRPC (grpcio).

Full guide: gRPC integration docs

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], validate=True)
container.open()  # or: with container: ... — required under modern-di 3.x's mandatory-open lifecycle
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 lifecycle is yours to own end-to-end: call .open() (or use with/async with) before constructing the interceptor and serving traffic — required under modern-di 3.x's mandatory-open lifecycle, since DIInterceptor/DIAioInterceptor never open the root themselves — and call close_sync() (or await close_async() on grpc.aio) after the server stops.

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.