Merge branch 'main' into asgi-support

faststream/broker/fastapi/route.py

@@ -18,6 +18,7 @@
 from starlette.requests import Request
 
 from faststream.broker.fastapi.get_dependant import get_fastapi_native_dependant
+from faststream.broker.response import Response, ensure_response
 from faststream.broker.types import P_HandlerParams, T_HandlerReturn
 
 from ._compat import (
@@ -186,15 +187,15 @@ def make_fastapi_execution(
     response_model_exclude_none: bool,
 ) -> Callable[
     ["StreamMessage", "NativeMessage[Any]"],
-    Awaitable[Any],
+    Awaitable[Response],
 ]:
     """Creates a FastAPI application."""
     is_coroutine = asyncio.iscoroutinefunction(dependent.call)
 
     async def app(
         request: "StreamMessage",
         raw_message: "NativeMessage[Any]",  # to support BackgroundTasks by middleware
-    ) -> Any:
+    ) -> Response:
         """Consume StreamMessage and return user function result."""
         async with AsyncExitStack() as stack:
             if FASTAPI_V106:
@@ -215,14 +216,16 @@ async def app(
             if solved_result.errors:
                 raise_fastapi_validation_error(solved_result.errors, request._body)  # type: ignore[arg-type]
 
-            raw_reponse = await run_endpoint_function(
+            function_result = await run_endpoint_function(
                 dependant=dependent,
                 values=solved_result.values,
                 is_coroutine=is_coroutine,
             )
 
-            content = await serialize_response(
-                response_content=raw_reponse,
+            response = ensure_response(function_result)
+
+            response.body = await serialize_response(
+                response_content=response.body,
                 field=response_field,
                 include=response_model_include,
                 exclude=response_model_exclude,
@@ -233,8 +236,8 @@ async def app(
                 is_coroutine=is_coroutine,
             )
 
-            return content
+            return response
 
-        return None
+        raise AssertionError("unreachable")
 
     return app

faststream/nats/response.py

@@ -1,4 +1,4 @@
-from typing import TYPE_CHECKING, Optional
+from typing import TYPE_CHECKING, Dict, Optional
 
 from typing_extensions import override
 
@@ -13,7 +13,7 @@ def __init__(
         self,
         body: "SendableMessage",
         *,
-        headers: Optional["AnyDict"] = None,
+        headers: Optional[Dict[str, str]] = None,
         correlation_id: Optional[str] = None,
         stream: Optional[str] = None,
     ) -> None:

faststream/nats/testing.py

@@ -18,7 +18,7 @@
 if TYPE_CHECKING:
     from faststream.nats.publisher.asyncapi import AsyncAPIPublisher
     from faststream.nats.subscriber.usecase import LogicSubscriber
-    from faststream.types import AnyDict, SendableMessage
+    from faststream.types import SendableMessage
 
 __all__ = ("TestNatsBroker",)
 
@@ -187,7 +187,7 @@ def build_message(
     *,
     reply_to: str = "",
     correlation_id: Optional[str] = None,
-    headers: Optional["AnyDict"] = None,
+    headers: Optional[Dict[str, str]] = None,
 ) -> "PatchedMessage":
     msg, content_type = encode_message(message)
     return PatchedMessage(

faststream/rabbit/publisher/usecase.py

@@ -97,7 +97,7 @@ class RequestPublishKwargs(TypedDict, total=False):
     ]
 
 
-class PublishKwargs(RequestPublishKwargs):
+class PublishKwargs(RequestPublishKwargs, total=False):
     """Typed dict to annotate RabbitMQ publishers."""
 
     reply_to: Annotated[
@@ -181,7 +181,7 @@ def __hash__(self) -> int:
         )
 
     @override
-    async def publish(  # type: ignore[override]
+    async def publish(
         self,
         message: "AioPikaSendableMessage",
         queue: Annotated[

pyproject.toml

@@ -88,10 +88,10 @@ devdocs = [
     "mdx-include==1.4.2",
     "mkdocstrings[python]==0.26.1",
     "mkdocs-literate-nav==0.6.1",
-    "mkdocs-git-revision-date-localized-plugin==1.2.8",
+    "mkdocs-git-revision-date-localized-plugin==1.2.9",
     "mike==2.1.3",                                      # versioning
     "mkdocs-minify-plugin==0.8.0",
-    "mkdocs-macros-plugin==1.0.5",                      # includes with variables
+    "mkdocs-macros-plugin==1.2.0",                      # includes with variables
     "mkdocs-glightbox==0.4.0",                          # img zoom
     "pillow",                                           # required for mkdocs-glightbo
     "cairosvg",                                         # required for mkdocs-glightbo
@@ -114,23 +114,23 @@ types = [
 
 lint = [
     "faststream[types]",
-    "ruff==0.6.4",
+    "ruff==0.6.5",
     "bandit==1.7.9",
-    "semgrep==1.86.0",
+    "semgrep==1.87.0",
     "codespell==2.3.0",
 ]
 
 test-core = [
     "coverage[toml]==7.6.1",
-    "pytest==8.3.2",
+    "pytest==8.3.3",
     "pytest-asyncio==0.24.0",
     "dirty-equals==0.8.0",
     "typing-extensions>=4.8.0,<4.12.1; python_version < '3.9'",  # to fix dirty-equals
 ]
 
 testing = [
     "faststream[test-core]",
-    "fastapi==0.114.0",
+    "fastapi==0.114.2",
     "pydantic-settings>=2.0.0,<3.0.0",
     "httpx==0.27.2",
     "PyYAML==6.0.2",

tests/brokers/base/fastapi.py

@@ -8,7 +8,7 @@
 from fastapi.exceptions import RequestValidationError
 from fastapi.testclient import TestClient
 
-from faststream import context
+from faststream import Response, context
 from faststream.broker.core.usecase import BrokerUsecase
 from faststream.broker.fastapi.context import Context
 from faststream.broker.fastapi.router import StreamRouter
@@ -226,6 +226,30 @@ async def hello():
                 )
                 assert r == "hi", r
 
+    async def test_request(self, queue: str):
+        """Local test due request exists in all TestClients."""
+        router = self.router_class(setup_state=False)
+
+        app = FastAPI()
+
+        args, kwargs = self.get_subscriber_params(queue)
+
+        @router.subscriber(*args, **kwargs)
+        async def hello():
+            return Response("Hi!", headers={"x-header": "test"})
+
+        async with self.broker_test(router.broker):
+            with TestClient(app) as client:
+                assert not client.app_state.get("broker")
+
+                r = await router.broker.request(
+                    "hi",
+                    queue,
+                    timeout=0.5,
+                )
+                assert await r.decode() == "Hi!"
+                assert r.headers["x-header"] == "test"
+
     async def test_base_without_state(self, queue: str):
         router = self.router_class(setup_state=False)