feat: add grpcio integration

Author: notypecheck

aioinject/ext/grpcio.py

@@ -0,0 +1,122 @@
+from __future__ import annotations
+
+import contextvars
+from typing import TYPE_CHECKING
+
+from grpc import (  # type: ignore[import-untyped]
+    HandlerCallDetails,
+    RpcMethodHandler,
+    ServicerContext,
+    stream_stream_rpc_method_handler,
+    stream_unary_rpc_method_handler,
+    unary_stream_rpc_method_handler,
+    unary_unary_rpc_method_handler,
+)
+from grpc.aio import ServerInterceptor  # type: ignore[import-untyped]
+
+from aioinject._types import (
+    P,
+    T,
+)
+from aioinject.decorators import base_inject
+
+
+if TYPE_CHECKING:
+    from collections.abc import AsyncIterator, Awaitable, Callable
+
+    from google.protobuf.message import Message  # type: ignore[import-untyped]
+
+    from aioinject import Container, Context
+
+__all__ = ["AioInjectInterceptor", "inject"]
+
+
+def inject(function: Callable[P, T]) -> Callable[P, T]:
+    return base_inject(
+        function,
+        context_parameters=(),
+        context_getter=lambda args, kwargs: _context_var.get(),  # noqa: ARG005
+    )
+
+
+_context_var: contextvars.ContextVar[Context] = contextvars.ContextVar(
+    "aioinject.grpcio.context"
+)
+
+
+class AioInjectInterceptor(ServerInterceptor):  # type: ignore[misc]
+    def __init__(self, container: Container) -> None:
+        self._container = container
+
+    async def intercept_service(  # noqa: C901
+        self,
+        continuation: Callable[
+            [HandlerCallDetails],
+            Awaitable[RpcMethodHandler],
+        ],
+        handler_call_details: HandlerCallDetails,
+    ) -> RpcMethodHandler:
+        handler = await continuation(handler_call_details)
+        deserializer = handler.request_deserializer
+        serializer = handler.response_serializer
+
+        if handler.unary_unary:
+
+            async def unary_unary_behavior(
+                request: Message, context: ServicerContext
+            ) -> object:
+                async with self._container.context() as di_context:
+                    _context_var.set(di_context)
+                    return await handler.unary_unary(request, context)
+
+            return unary_unary_rpc_method_handler(
+                behavior=unary_unary_behavior,
+                request_deserializer=deserializer,
+                response_serializer=serializer,
+            )
+        if handler.unary_stream:
+
+            async def unary_stream_behavior(
+                request: Message, context: ServicerContext
+            ) -> AsyncIterator[object]:
+                async with self._container.context() as di_context:
+                    _context_var.set(di_context)
+                    async for message in handler.unary_stream(
+                        request, context
+                    ):
+                        yield message
+
+            return unary_stream_rpc_method_handler(
+                unary_stream_behavior,
+                request_deserializer=deserializer,
+                response_serializer=serializer,
+            )
+        if handler.stream_unary:
+
+            async def stream_unary_behavior(
+                request: AsyncIterator[Message], context: ServicerContext
+            ) -> object:
+                _context_var.set(self._container.root)
+                return await handler.stream_unary(request, context)
+
+            return stream_unary_rpc_method_handler(
+                stream_unary_behavior,
+                request_deserializer=deserializer,
+                response_serializer=serializer,
+            )
+
+        if handler.stream_stream:
+
+            async def stream_stream_behavior(
+                request: AsyncIterator[Message], context: ServicerContext
+            ) -> AsyncIterator[object]:
+                _context_var.set(self._container.root)
+                async for message in handler.stream_stream(request, context):
+                    yield message
+
+            return stream_stream_rpc_method_handler(
+                stream_stream_behavior,
+                request_deserializer=deserializer,
+                response_serializer=serializer,
+            )
+        return handler  # pragma: no cover

pyproject.toml

@@ -32,26 +32,28 @@ benchmark = [
 ]
 dev = [
     "mypy>=1.17.1",
-    "ruff>=0.12.9",
+    "ruff>=0.12.10",
 ]
 docs = [
     "mkdocs>=1.6.1",
-    "mkdocs-material>=9.6.17",
+    "mkdocs-material>=9.6.18",
 ]
 test = [
     "aiogram>=3.22.0",
     "anyio>=4.10.0",
     "asgi-lifespan>=2.1.0",
-    "coverage[toml]>=7.10.4",
+    "coverage[toml]>=7.10.5",
     "django>=5.2.5",
     "djangorestframework>=3.16.1",
     "fastapi>=0.116.1",
+    "grpcio>=1.75.0",
+    "grpcio-tools>=1.75.0",
     "litestar>=2.17.0",
     "pydantic-settings>=2.10.1",
     "pytest>=8.4.1",
     "pytest-cov>=6.2.1",
     "pytest-django>=4.11.1",
-    "strawberry-graphql>=0.278.1",
+    "strawberry-graphql>=0.280.0",
     "trio>=0.30.0",
     "uvicorn>=0.35.0",
 ]
@@ -152,6 +154,7 @@ ignore = [
     "S101",
     "S311", # random
 ]
+"tests/integrations/grpcio/*" = ["ARG002", "N802"]
 "tests/*/test_*.py" = ["FBT001"]
 "docs/*" = [
     "INP001", # Implicit package (no __init__.py)

tests/integrations/conftest.py

@@ -5,6 +5,8 @@
 import pytest
 
 import aioinject
+from aioinject import Context, FromContext
+from aioinject.scope import CurrentScope
 
 
 @pytest.fixture
@@ -33,6 +35,7 @@ def get_node() -> ScopedNode:
 @pytest.fixture
 def container(provided_value: int) -> aioinject.Container:
     container = aioinject.Container()
+    container.register(FromContext(Context, scope=CurrentScope()))
     container.register(aioinject.Object(provided_value))
     container.register(aioinject.Scoped(NumberService))
     container.register(aioinject.Scoped(get_node))

tests/integrations/grpcio/__init__.py

@@ -0,0 +1,16 @@
+syntax = "proto3";
+
+message Request {
+    string field = 1;
+}
+
+message Response {
+    string field = 1;
+}
+
+service Service {
+    rpc Unary(Request) returns (Response);
+    rpc UnaryStream(Request) returns (stream Response);
+    rpc StreamUnary(stream Request) returns (Response);
+    rpc StreamStream(stream Request) returns (stream Response);
+}

tests/integrations/grpcio/service.proto

@@ -0,0 +1,147 @@
+import uuid
+from collections.abc import AsyncIterator
+from pathlib import Path
+from typing import Any, cast
+
+import grpc  # type: ignore[import-untyped]
+import pytest
+from _pytest.fixtures import SubRequest
+from grpc import ServicerContext
+from grpc.aio import Server  # type: ignore[import-untyped]
+
+from aioinject import Container, Context, Injected, Scope
+from aioinject.ext.grpcio import AioInjectInterceptor, inject
+
+
+protos: Any
+services: Any
+protos, services = grpc.protos_and_services(
+    str(
+        Path(__file__).parent.joinpath("service.proto").relative_to(Path.cwd())
+    )
+)
+
+
+@pytest.fixture(scope="session", autouse=True, params=["asyncio"])
+def anyio_backend(request: SubRequest) -> str:
+    return cast("str", request.param)
+
+
+class Service(services.Service):  # type: ignore[misc]
+    @inject
+    async def Unary(
+        self,
+        request: protos.Request,
+        context: ServicerContext,
+        value: Injected[int],
+        di_context: Injected[Context],
+    ) -> protos.Response:
+        assert di_context.scope is Scope.request
+        return protos.Response(field=f"{request.field} {value}")
+
+    @inject
+    async def UnaryStream(
+        self,
+        request: protos.Request,
+        context: ServicerContext,
+        value: Injected[int],
+        di_context: Injected[Context],
+    ) -> AsyncIterator[protos.Response]:
+        assert di_context.scope is Scope.request
+
+        for i in range(10):
+            yield protos.Response(field=f"{request.field} {value} {i}")
+
+    @inject
+    async def StreamUnary(
+        self,
+        request: AsyncIterator[protos.Request],
+        context: ServicerContext,
+        value: Injected[int],
+        di_context: Injected[Context],
+    ) -> protos.Response:
+        assert di_context.scope == Scope.lifetime
+
+        values = [message.field async for message in request]
+        values.append(str(value))
+        return protos.Response(field=" ".join(values))
+
+    @inject
+    async def StreamStream(
+        self,
+        request: AsyncIterator[protos.Request],
+        context: ServicerContext,
+        value: Injected[int],
+        di_context: Injected[Context],
+    ) -> AsyncIterator[protos.Response]:
+        assert di_context.scope == Scope.lifetime
+
+        async for message in request:
+            yield protos.Response(field=f"{message.field} {value}")
+
+
+@pytest.fixture
+async def grpcio_server(container: Container) -> AsyncIterator[Server]:
+    server = grpc.aio.server(interceptors=[AioInjectInterceptor(container)])
+    services.add_ServiceServicer_to_server(Service(), server)
+    server.add_insecure_port("localhost:50051")
+    await server.start()
+    yield server
+    await server.stop(0)
+
+
+@pytest.fixture
+async def grpcio_client(
+    grpcio_server: object,  # noqa: ARG001
+) -> AsyncIterator[services.ServiceStub]:
+    async with (
+        grpc.aio.insecure_channel("localhost:50051") as channel,
+    ):
+        yield services.ServiceStub(channel)
+
+
+async def test_unary_unary_ok(
+    grpcio_client: services.ServiceStub, provided_value: int
+) -> None:
+    field = str(uuid.uuid4())
+    response = await grpcio_client.Unary(protos.Request(field=field))
+    assert response.field == f"{field} {provided_value}"
+
+
+async def test_unary_stream_ok(
+    grpcio_client: services.ServiceStub, provided_value: int
+) -> None:
+    field = str(uuid.uuid4())
+    messages = [
+        message
+        async for message in grpcio_client.UnaryStream(
+            protos.Request(field=field)
+        )
+    ]
+    for number, message in enumerate(messages):
+        assert message.field == f"{field} {provided_value} {number}"
+
+
+async def test_stream_unary_ok(
+    grpcio_client: services.ServiceStub, provided_value: int
+) -> None:
+    fields = [str(uuid.uuid4()) for _ in range(10)]
+    call = grpcio_client.StreamUnary()
+    for field in fields:
+        await call.write(protos.Request(field=field))
+    await call.done_writing()
+
+    response = await call
+    assert response.field == " ".join([*fields, str(provided_value)])
+
+
+async def test_stream_stream_ok(
+    grpcio_client: services.ServiceStub,
+    provided_value: int,
+) -> None:
+    call = grpcio_client.StreamStream()
+    for _ in range(10):
+        field = str(uuid.uuid4())
+        await call.write(protos.Request(field=field))
+        response = await call.read()
+        assert response.field == f"{field} {provided_value}"