feat: asgi request

faststream/_internal/application.py

@@ -84,6 +84,7 @@ def _init_setupable_(  # noqa: PLW3201
         config: Optional["FastDependsConfig"] = None,
     ) -> None:
         self.config = config or FastDependsConfig()
+        self.config.context.set_global("app", self)
 
         self.brokers = [broker] if broker else []
 
@@ -128,12 +129,10 @@ def __init__(
         after_shutdown: Sequence["AnyCallable"] = (),
         specification: Optional["SpecificationFactory"] = None,
     ) -> None:
-        super().__init__(broker, config=config, specification=specification)
-
-        self.context.set_global("app", self)
-
         self.logger = logger
 
+        super().__init__(broker, config=config, specification=specification)
+
         self._on_startup_calling: list[AsyncFunc] = [
             apply_types(
                 to_async(x),

faststream/asgi/__init__.py

@@ -1,13 +1,20 @@
+from .annotations import Request
 from .app import AsgiFastStream
 from .factories import AsyncAPIRoute, make_asyncapi_asgi, make_ping_asgi
-from .handlers import get
+from .handlers import HttpHandler, get, post
+from .params import Header, Query
 from .response import AsgiResponse
 
 __all__ = (
     "AsgiFastStream",
     "AsgiResponse",
     "AsyncAPIRoute",
+    "Header",
+    "HttpHandler",
+    "Query",
+    "Request",
     "get",
     "make_asyncapi_asgi",
     "make_ping_asgi",
+    "post",
 )

faststream/asgi/annotations.py

@@ -0,0 +1,9 @@
+from typing import Annotated
+
+from faststream._internal.context import Context
+
+from .request import AsgiRequest
+
+__all__ = ("Request",)
+
+Request = Annotated[AsgiRequest, Context("request")]

faststream/asgi/app.py

@@ -103,6 +103,8 @@ def __init__(
         specification: Optional["SpecificationFactory"] = None,
         asyncapi_path: str | AsyncAPIRoute | None = None,
     ) -> None:
+        self.routes = list(asgi_routes)
+
         super().__init__(
             broker,
             logger=logger,
@@ -119,20 +121,26 @@ def __init__(
             specification=specification,
         )
 
-        self.routes = list(asgi_routes)
         if asyncapi_path:
-            route = AsyncAPIRoute.ensure_route(asyncapi_path)
-            self.routes.append((route.path, route(self.schema)))
-
-        for path, app in self.routes:
-            if isinstance(app, HttpHandler):
-                self.schema.add_http_route(path, app)
+            asyncapi_route = AsyncAPIRoute.ensure_route(asyncapi_path)
+            self.routes.append((asyncapi_route.path, asyncapi_route(self.schema)))
 
         self._server = OuterRunState()
 
         self._log_level: int = logging.INFO
         self._run_extra_options: dict[str, SettingField] = {}
 
+    def _init_setupable_(  # noqa: PLW3201
+        self,
+        broker: Optional["BrokerUsecase[Any, Any]"] = None,
+        /,
+        specification: Optional["SpecificationFactory"] = None,
+        config: Optional["FastDependsConfig"] = None,
+    ) -> None:
+        super()._init_setupable_(broker, specification, config)
+        for route in self.routes:
+            self._register_route(route)
+
     @classmethod
     def from_app(
         cls,
@@ -155,7 +163,16 @@ def from_app(
         return asgi_app
 
     def mount(self, path: str, route: "ASGIApp") -> None:
-        self.routes.append((path, route))
+        asgi_route = (path, route)
+        self.routes.append(asgi_route)
+        self._register_route(asgi_route)
+
+    def _register_route(self, asgi_route: tuple[str, "ASGIApp"]) -> None:
+        path, route = asgi_route
+        if isinstance(route, HttpHandler):
+            self.schema.add_http_route(path, route)
+            route.update_fd_config(self.config)
+            route.set_logger(self.logger)
 
     async def __call__(
         self,

faststream/asgi/factories.py

@@ -9,14 +9,13 @@
 
 from .handlers import get
 from .response import AsgiResponse
+from .types import ASGIApp, Scope
 
 if TYPE_CHECKING:
     from faststream._internal.broker import BrokerUsecase
     from faststream.specification.base import SpecificationFactory
     from faststream.specification.schema import Tag, TagDict
 
-    from .types import ASGIApp, Scope
-
 
 def make_ping_asgi(
     broker: "BrokerUsecase[Any, Any]",
@@ -26,7 +25,7 @@ def make_ping_asgi(
     description: str | None = None,
     tags: Sequence[Union["Tag", "TagDict", dict[str, Any]]] | None = None,
     unique_id: str | None = None,
-) -> "ASGIApp":
+) -> ASGIApp:
     healthy_response = AsgiResponse(b"", 204)
     unhealthy_response = AsgiResponse(b"", 500)
 
@@ -36,7 +35,7 @@ def make_ping_asgi(
         tags=tags,
         unique_id=unique_id,
     )
-    async def ping(scope: "Scope") -> AsgiResponse:
+    async def ping(scope: Scope) -> AsgiResponse:
         if await broker.ping(timeout):
             return healthy_response
         return unhealthy_response
@@ -125,7 +124,7 @@ def make_asyncapi_asgi(
     expand_message_examples: bool = True,
     asyncapi_js_url: str = ASYNCAPI_JS_DEFAULT_URL,
     asyncapi_css_url: str = ASYNCAPI_CSS_DEFAULT_URL,
-) -> "ASGIApp":
+) -> ASGIApp:
     cached_docs = None
 
     @get(
@@ -134,7 +133,7 @@ def make_asyncapi_asgi(
         tags=tags,
         unique_id=unique_id,
     )
-    async def docs(scope: "Scope") -> AsgiResponse:
+    async def docs(scope: Scope) -> AsgiResponse:
         nonlocal cached_docs
         if not cached_docs:
             cached_docs = get_asyncapi_html(

faststream/asgi/handlers.py

@@ -1,15 +1,25 @@
+import logging
 from collections.abc import Callable, Sequence
 from typing import TYPE_CHECKING, Any, Optional, Union, overload
 
+from faststream import apply_types
+from faststream._internal.context.repository import ContextRepo
+from faststream._internal.di.config import FastDependsConfig
+from faststream._internal.utils.functions import to_async
+
+from .request import AsgiRequest
 from .response import AsgiResponse
 
 if TYPE_CHECKING:
+    from faststream._internal.basic_types import LoggerProto
     from faststream.specification.schema import Tag, TagDict
 
     from .types import ASGIApp, Receive, Scope, Send, UserApp
 
 
 class HttpHandler:
+    _context_repo: ContextRepo
+
     def __init__(
         self,
         func: "UserApp",
@@ -19,26 +29,55 @@ def __init__(
         methods: Sequence[str] | None = None,
         tags: Sequence[Union["Tag", "TagDict", dict[str, Any]]] | None = None,
         unique_id: str | None = None,
+        fd_config: FastDependsConfig | None = None,
     ) -> None:
+        self.__original_func = func
         self.func = func
         self.methods = methods or ()
         self.include_in_schema = include_in_schema
         self.description = description or func.__doc__
         self.tags = tags
         self.unique_id = unique_id
+        self.fd_config = fd_config or FastDependsConfig()
+        self.logger: LoggerProto | None = None
 
     async def __call__(self, scope: "Scope", receive: "Receive", send: "Send") -> None:
         if scope["method"] not in self.methods:
             response: ASGIApp = _get_method_not_allowed_response(self.methods)
 
         else:
-            try:
-                response = await self.func(scope)
-            except Exception:
-                response = AsgiResponse(body=b"Internal Server Error", status_code=500)
-
+            with (
+                self.fd_config.context.scope(
+                    "request", AsgiRequest(scope, receive, send)
+                ),
+                self.fd_config.context.scope(
+                    "logger",
+                    self.logger,
+                ),
+            ):
+                try:
+                    response = await self.func(scope)
+                except Exception:
+                    if self.logger is not None:
+                        self.logger.log(
+                            logging.ERROR,
+                            "Exception occurred while processing request",
+                            exc_info=True,
+                        )
+                    response = AsgiResponse(
+                        body=b"Internal Server Error", status_code=500
+                    )
         await response(scope, receive, send)
 
+    def update_fd_config(self, config: FastDependsConfig) -> None:
+        self.fd_config = config | self.fd_config
+        self.func = apply_types(
+            to_async(self.__original_func), context__=self.fd_config.context
+        )
+
+    def set_logger(self, logger: "LoggerProto | None") -> None:
+        self.logger = logger
+
 
 class GetHandler(HttpHandler):
     def __init__(
@@ -105,6 +144,71 @@ def decorator(inner_func: "UserApp") -> "ASGIApp":
     return decorator(func)
 
 
+class PostHandler(HttpHandler):
+    def __init__(
+        self,
+        func: "UserApp",
+        *,
+        include_in_schema: bool = True,
+        description: str | None = None,
+        tags: Sequence[Union["Tag", "TagDict", dict[str, Any]]] | None = None,
+        unique_id: str | None = None,
+    ) -> None:
+        super().__init__(
+            func,
+            include_in_schema=include_in_schema,
+            description=description,
+            methods=("POST", "HEAD"),
+            tags=tags,
+            unique_id=unique_id,
+        )
+
+
+@overload
+def post(
+    func: "UserApp",
+    *,
+    include_in_schema: bool = True,
+    description: str | None = None,
+    tags: Sequence[Union["Tag", "TagDict", dict[str, Any]]] | None = None,
+    unique_id: str | None = None,
+) -> "ASGIApp": ...
+
+
+@overload
+def post(
+    func: None = None,
+    *,
+    include_in_schema: bool = True,
+    description: str | None = None,
+    tags: Sequence[Union["Tag", "TagDict", dict[str, Any]]] | None = None,
+    unique_id: str | None = None,
+) -> Callable[["UserApp"], "ASGIApp"]: ...
+
+
+def post(
+    func: Optional["UserApp"] = None,
+    *,
+    include_in_schema: bool = True,
+    description: str | None = None,
+    tags: Sequence[Union["Tag", "TagDict", dict[str, Any]]] | None = None,
+    unique_id: str | None = None,
+) -> Union[Callable[["UserApp"], "ASGIApp"], "ASGIApp"]:
+    def decorator(inner_func: "UserApp") -> "ASGIApp":
+        return PostHandler(
+            inner_func,
+            include_in_schema=include_in_schema,
+            description=description,
+            tags=tags,
+            unique_id=unique_id,
+        )
+
+    if func is None:
+        return decorator
+
+    return decorator(func)
+
+
 def _get_method_not_allowed_response(methods: Sequence[str]) -> AsgiResponse:
     return AsgiResponse(
         body=b"Method Not Allowed",

faststream/asgi/params.py

@@ -0,0 +1,32 @@
+from typing import Any
+
+from faststream._internal.constants import EMPTY
+from faststream._internal.context import Context as Context_
+
+
+def Header(  # noqa: N802
+    real_name: str = "",
+    *,
+    cast: bool = False,
+    default: Any = EMPTY,
+) -> Any:
+    return Context_(
+        real_name=real_name.lower(),
+        cast=cast,
+        default=default,
+        prefix="request.headers.",
+    )
+
+
+def Query(  # noqa: N802
+    real_name: str = "",
+    *,
+    cast: bool = False,
+    default: Any = EMPTY,
+) -> Any:
+    return Context_(
+        real_name=real_name,
+        cast=cast,
+        default=default,
+        prefix="request.query_params.",
+    )