docs: add FastStream integration

docs/index.md

@@ -9,6 +9,7 @@
     introduction/ioc-container
     introduction/fastapi
     introduction/litestar
+    introduction/faststream
     introduction/inject-factories
     introduction/multiple-containers
     introduction/dynamic-container

docs/introduction/fastapi.md

@@ -32,7 +32,7 @@ async def read_root(
         container.DependentFactory,
         fastapi.Depends(container.DIContainer.dependent_factory),
     ],
-) -> datetime.datetime:
+) -> str:
     return some_dependency.async_resource
 
 

docs/introduction/faststream.md

@@ -0,0 +1,42 @@
+# Usage with `FastStream`
+
+```python
+import datetime
+import contextlib
+import typing
+
+from faststream import FastStream, Depends, Logger
+from faststream.rabbit import RabbitBroker
+
+from tests import container
+
+
+@contextlib.asynccontextmanager
+async def lifespan_manager() -> typing.AsyncIterator[None]:
+    try:
+        yield
+    finally:
+        await container.DIContainer.tear_down()
+
+
+broker = RabbitBroker()
+app = FastStream(broker, lifespan=lifespan_manager)
+
+
+@broker.subscriber("in")
+async def read_root(
+    logger: Logger,
+    some_dependency: typing.Annotated[
+        container.DependentFactory,
+        Depends(container.DIContainer.dependent_factory)
+    ],
+) -> datetime.datetime:
+    startup_time = some_dependency.async_resource
+    logger.info(startup_time)
+    return startup_time
+
+
+@app.after_startup
+async def t() -> None:
+    await broker.publish(None, "in")
+```

pyproject.toml

@@ -27,8 +27,6 @@ docs = "https://that-depends.readthedocs.io"
 
 [dependency-groups]
 dev = [
-    "fastapi",
-    "litestar",
     "httpx",
     "pytest",
     "pytest-cov",
@@ -37,7 +35,11 @@ dev = [
     "ruff",
     "mypy",
     "typing-extensions",
-    "pre-commit"
+    "pre-commit",
+    # integrations tests
+    "fastapi",
+    "litestar",
+    "faststream[nats]",
 ]
 
 [build-system]

tests/integrations/faststream/__init__.py

@@ -0,0 +1,5 @@
+import pytest
+
+
+pytest.importorskip("faststream")
+pytest.importorskip("nats")

tests/integrations/faststream/test_faststream_di.py

@@ -0,0 +1,46 @@
+import datetime
+import typing
+
+from faststream import Depends
+from faststream.nats import NatsBroker, TestNatsBroker
+
+from tests import container
+
+
+broker = NatsBroker()
+
+TEST_SUBJECT = "test"
+
+
+@broker.subscriber(TEST_SUBJECT)
+async def index_subscruber(
+    dependency: typing.Annotated[
+        container.DependentFactory,
+        Depends(container.DIContainer.dependent_factory),
+    ],
+    free_dependency: typing.Annotated[
+        container.FreeFactory,
+        Depends(container.DIContainer.resolver(container.FreeFactory)),
+    ],
+    singleton: typing.Annotated[
+        container.SingletonFactory,
+        Depends(container.DIContainer.singleton),
+    ],
+    singleton_attribute: typing.Annotated[bool, Depends(container.DIContainer.singleton.dep1)],
+) -> datetime.datetime:
+    assert dependency.sync_resource == free_dependency.dependent_factory.sync_resource
+    assert dependency.async_resource == free_dependency.dependent_factory.async_resource
+    assert singleton.dep1 is True
+    assert singleton_attribute is True
+    return dependency.async_resource
+
+
+async def test_read_main() -> None:
+    async with TestNatsBroker(broker) as br:
+        result = await br.request(None, TEST_SUBJECT)
+
+        result_str = typing.cast(str, await result.decode())
+        assert (
+            datetime.datetime.fromisoformat(result_str.replace("Z", "+00:00"))
+            == await container.DIContainer.async_resource()
+        )

tests/integrations/faststream/test_faststream_di_pass_message.py

@@ -0,0 +1,50 @@
+import typing
+
+from faststream import BaseMiddleware, Context, Depends
+from faststream.broker.message import StreamMessage
+from faststream.nats import NatsBroker, TestNatsBroker
+from faststream.nats.message import NatsMessage
+
+from that_depends import BaseContainer, container_context, fetch_context_item, providers
+
+
+class ContextMiddleware(BaseMiddleware):
+    async def consume_scope(
+        self,
+        call_next: typing.Callable[..., typing.Awaitable[typing.Any]],
+        msg: StreamMessage[typing.Any],
+    ) -> typing.Any:  # noqa: ANN401
+        async with container_context({"request": msg}):
+            return await call_next(msg)
+
+
+broker = NatsBroker(middlewares=(ContextMiddleware,), validate=False)
+
+TEST_SUBJECT = "test"
+
+
+class DIContainer(BaseContainer):
+    context_request = providers.Factory(
+        lambda: fetch_context_item("request"),
+    )
+
+
+@broker.subscriber(TEST_SUBJECT)
+async def index_subscruber(
+    context_request: typing.Annotated[
+        NatsMessage,
+        Depends(DIContainer.context_request, cast=False),
+    ],
+    message: typing.Annotated[
+        NatsMessage,
+        Context(),
+    ],
+) -> bool:
+    return message is context_request
+
+
+async def test_read_main() -> None:
+    async with TestNatsBroker(broker) as br:
+        result = await br.request(None, TEST_SUBJECT)
+
+        assert (await result.decode()) is True