diff --git a/faststream/_internal/application.py b/faststream/_internal/application.py index 96ceaceae66..5aaf927072d 100644 --- a/faststream/_internal/application.py +++ b/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), diff --git a/faststream/asgi/__init__.py b/faststream/asgi/__init__.py index bcd177da593..eba65e61cf8 100644 --- a/faststream/asgi/__init__.py +++ b/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", ) diff --git a/faststream/asgi/annotations.py b/faststream/asgi/annotations.py new file mode 100644 index 00000000000..3eb574c1ee3 --- /dev/null +++ b/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")] diff --git a/faststream/asgi/app.py b/faststream/asgi/app.py index f5f16614fa9..fde18a84f30 100644 --- a/faststream/asgi/app.py +++ b/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, diff --git a/faststream/asgi/factories.py b/faststream/asgi/factories.py index 45895e29fcb..4eb56ec9899 100644 --- a/faststream/asgi/factories.py +++ b/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( diff --git a/faststream/asgi/handlers.py b/faststream/asgi/handlers.py index 329f91d78a4..c5b8fbd6827 100644 --- a/faststream/asgi/handlers.py +++ b/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", diff --git a/faststream/asgi/params.py b/faststream/asgi/params.py new file mode 100644 index 00000000000..c35e37d44f5 --- /dev/null +++ b/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.", + ) diff --git a/faststream/asgi/request.py b/faststream/asgi/request.py new file mode 100644 index 00000000000..085b057fd26 --- /dev/null +++ b/faststream/asgi/request.py @@ -0,0 +1,80 @@ +import json +from collections.abc import AsyncGenerator +from typing import TYPE_CHECKING, Any, cast +from urllib.parse import parse_qs + +from faststream._internal.constants import EMPTY + +if TYPE_CHECKING: + from .types import Receive, Scope, Send + + +class ClientDisconnectError(Exception): ... + + +class AsgiRequest: + def __init__(self, scope: "Scope", receive: "Receive", send: "Send") -> None: + self._scope = scope + self._receive = receive + self._send = send + self._stream_consumed = False + self._body: bytes | None = None + self._json: Any = EMPTY + self._query_params: dict[str, list[str]] | None = None + self._headers: dict[str, str] | None = None + + async def stream(self) -> AsyncGenerator[bytes, None]: + if self._body is not None: + yield self._body + yield b"" + return + if self._stream_consumed: + msg = "Stream consumed" + raise RuntimeError(msg) + while not self._stream_consumed: + message = await self._receive() + if message["type"] == "http.request": + body = message.get("body", b"") + if not message.get("more_body", False): + self._stream_consumed = True + if body: + yield body + elif message["type"] == "http.disconnect": # pragma: no branch + self._is_disconnected = True + raise ClientDisconnectError + yield b"" + + async def body(self) -> bytes: + if self._body is None: + chunks: list[bytes] = [chunk async for chunk in self.stream()] + self._body = b"".join(chunks) + return self._body + + async def json(self) -> Any: + if self._json is EMPTY: + body = await self.body() + self._json = json.loads(body) + return self._json + + @property + def query_params(self) -> dict[str, list[str]]: + if self._query_params is None: + query_string = self._scope.get("query_string", b"") + self._query_params = { + key.decode("latin-1"): [v.decode("latin-1") for v in value] + for key, value in parse_qs(query_string).items() + } + return self._query_params + + @property + def headers(self) -> dict[str, str]: + if self._headers is None: + self._headers = { + key.lower().decode("latin-1"): value.decode("latin-1") + for key, value in list(self._scope["headers"]) + } + return self._headers + + @property + def method(self) -> str: + return cast("str", self._scope["method"]) diff --git a/faststream/asgi/types.py b/faststream/asgi/types.py index d64c7188213..22d55e7eed1 100644 --- a/faststream/asgi/types.py +++ b/faststream/asgi/types.py @@ -6,4 +6,4 @@ Receive = Callable[[], Awaitable[Message]] Send = Callable[[Message], Awaitable[None]] ASGIApp = Callable[[Scope, Receive, Send], Awaitable[None]] -UserApp = Callable[[Scope], Awaitable[ASGIApp]] +UserApp = Callable[..., Awaitable[ASGIApp]] diff --git a/tests/asgi/testcase.py b/tests/asgi/testcase.py index 1eb41e2618a..552765f9b1a 100644 --- a/tests/asgi/testcase.py +++ b/tests/asgi/testcase.py @@ -1,20 +1,25 @@ +from collections.abc import Callable from typing import Any from unittest.mock import AsyncMock import pytest +from fast_depends import Depends from starlette.applications import Starlette from starlette.routing import Mount from starlette.testclient import TestClient from starlette.websockets import WebSocketDisconnect +from faststream.annotations import FastStream, Logger from faststream.asgi import ( AsgiFastStream, AsgiResponse, + Request, get, make_asyncapi_asgi, make_ping_asgi, + post, ) -from faststream.asgi.types import Scope +from faststream.asgi.types import ASGIApp, Scope from faststream.specification import AsyncAPI @@ -96,8 +101,17 @@ async def test_asyncapi_asgi(self) -> None: assert response.text @pytest.mark.asyncio() - async def test_get_decorator(self) -> None: - @get + @pytest.mark.parametrize( + ("decorator", "client_method"), + ( + pytest.param(get, "get", id="get"), + pytest.param(post, "post", id="post"), + ), + ) + async def test_decorators( + self, decorator: Callable[..., ASGIApp], client_method: str + ) -> None: + @decorator async def some_handler(scope: Scope) -> AsgiResponse: return AsgiResponse(body=b"test", status_code=200) @@ -106,7 +120,63 @@ async def some_handler(scope: Scope) -> AsgiResponse: async with self.get_test_broker(broker): with TestClient(app) as client: - response = client.get("/test") + response = getattr(client, client_method)("/test") + assert response.status_code == 200 + assert response.text == "test" + + @pytest.mark.asyncio() + @pytest.mark.parametrize( + ("decorator", "client_method"), + ( + pytest.param(get, "get", id="get"), + pytest.param(post, "post", id="post"), + ), + ) + async def test_context_injected( + self, decorator: Callable[..., ASGIApp], client_method: str + ) -> None: + @decorator + async def some_handler( + request: Request, logger: Logger, app: FastStream + ) -> AsgiResponse: + return AsgiResponse( + body=f"{request.__class__.__name__} {logger.__class__.__name__} {app.__class__.__name__}".encode(), + status_code=200, + ) + + broker = self.get_broker() + app = AsgiFastStream(broker, asgi_routes=[("/test", some_handler)]) + + async with self.get_test_broker(broker): + with TestClient(app) as client: + response = getattr(client, client_method)("/test") + assert response.status_code == 200 + assert response.text == "AsgiRequest Logger AsgiFastStream" + + @pytest.mark.asyncio() + @pytest.mark.parametrize( + ("decorator", "client_method"), + ( + pytest.param(get, "get", id="get"), + pytest.param(post, "post", id="post"), + ), + ) + async def test_fast_depends_injected( + self, decorator: Callable[..., ASGIApp], client_method: str + ) -> None: + def get_string() -> str: + return "test" + + @decorator + async def some_handler(string=Depends(get_string)) -> AsgiResponse: # noqa: B008 + return AsgiResponse(body=string.encode(), status_code=200) + + broker = self.get_broker() + app = AsgiFastStream(broker, asgi_routes=[("/test", some_handler)]) + + async with self.get_test_broker(broker): + with TestClient(app) as client: + response = getattr(client, client_method)("/test") assert response.status_code == 200 assert response.text == "test"