diff options
Diffstat (limited to 'venv/lib/python3.11/site-packages/litestar/connection/base.py')
| -rw-r--r-- | venv/lib/python3.11/site-packages/litestar/connection/base.py | 345 |
1 files changed, 345 insertions, 0 deletions
diff --git a/venv/lib/python3.11/site-packages/litestar/connection/base.py b/venv/lib/python3.11/site-packages/litestar/connection/base.py new file mode 100644 index 0000000..d14c662 --- /dev/null +++ b/venv/lib/python3.11/site-packages/litestar/connection/base.py @@ -0,0 +1,345 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING, Any, Generic, TypeVar, cast + +from litestar._parsers import parse_cookie_string, parse_query_string +from litestar.datastructures.headers import Headers +from litestar.datastructures.multi_dicts import MultiDict +from litestar.datastructures.state import State +from litestar.datastructures.url import URL, Address, make_absolute_url +from litestar.exceptions import ImproperlyConfiguredException +from litestar.types.empty import Empty +from litestar.utils.empty import value_or_default +from litestar.utils.scope.state import ScopeState + +if TYPE_CHECKING: + from typing import NoReturn + + from litestar.app import Litestar + from litestar.types import DataContainerType, EmptyType + from litestar.types.asgi_types import Message, Receive, Scope, Send + from litestar.types.protocols import Logger + +__all__ = ("ASGIConnection", "empty_receive", "empty_send") + +UserT = TypeVar("UserT") +AuthT = TypeVar("AuthT") +HandlerT = TypeVar("HandlerT") +StateT = TypeVar("StateT", bound=State) + + +async def empty_receive() -> NoReturn: # pragma: no cover + """Raise a ``RuntimeError``. + + Serves as a placeholder ``send`` function. + + Raises: + RuntimeError + """ + raise RuntimeError() + + +async def empty_send(_: Message) -> NoReturn: # pragma: no cover + """Raise a ``RuntimeError``. + + Serves as a placeholder ``send`` function. + + Args: + _: An ASGI message + + Raises: + RuntimeError + """ + raise RuntimeError() + + +class ASGIConnection(Generic[HandlerT, UserT, AuthT, StateT]): + """The base ASGI connection container.""" + + __slots__ = ( + "scope", + "receive", + "send", + "_base_url", + "_url", + "_parsed_query", + "_cookies", + "_server_extensions", + "_connection_state", + ) + + scope: Scope + """The ASGI scope attached to the connection.""" + receive: Receive + """The ASGI receive function.""" + send: Send + """The ASGI send function.""" + + def __init__(self, scope: Scope, receive: Receive = empty_receive, send: Send = empty_send) -> None: + """Initialize ``ASGIConnection``. + + Args: + scope: The ASGI connection scope. + receive: The ASGI receive function. + send: The ASGI send function. + """ + self.scope = scope + self.receive = receive + self.send = send + self._connection_state = ScopeState.from_scope(scope) + self._base_url: URL | EmptyType = Empty + self._url: URL | EmptyType = Empty + self._parsed_query: tuple[tuple[str, str], ...] | EmptyType = Empty + self._cookies: dict[str, str] | EmptyType = Empty + self._server_extensions = scope.get("extensions") or {} # extensions may be None + + @property + def app(self) -> Litestar: + """Return the ``app`` for this connection. + + Returns: + The :class:`Litestar <litestar.app.Litestar>` application instance + """ + return self.scope["app"] + + @property + def route_handler(self) -> HandlerT: + """Return the ``route_handler`` for this connection. + + Returns: + The target route handler instance. + """ + return cast("HandlerT", self.scope["route_handler"]) + + @property + def state(self) -> StateT: + """Return the ``State`` of this connection. + + Returns: + A State instance constructed from the scope["state"] value. + """ + return cast("StateT", State(self.scope.get("state"))) + + @property + def url(self) -> URL: + """Return the URL of this connection's ``Scope``. + + Returns: + A URL instance constructed from the request's scope. + """ + if self._url is Empty: + if (url := self._connection_state.url) is not Empty: + self._url = url + else: + self._connection_state.url = self._url = URL.from_scope(self.scope) + + return self._url + + @property + def base_url(self) -> URL: + """Return the base URL of this connection's ``Scope``. + + Returns: + A URL instance constructed from the request's scope, representing only the base part + (host + domain + prefix) of the request. + """ + if self._base_url is Empty: + if (base_url := self._connection_state.base_url) is not Empty: + self._base_url = base_url + else: + scope = cast( + "Scope", + { + **self.scope, + "path": "/", + "query_string": b"", + "root_path": self.scope.get("app_root_path") or self.scope.get("root_path", ""), + }, + ) + self._connection_state.base_url = self._base_url = URL.from_scope(scope) + return self._base_url + + @property + def headers(self) -> Headers: + """Return the headers of this connection's ``Scope``. + + Returns: + A Headers instance with the request's scope["headers"] value. + """ + return Headers.from_scope(self.scope) + + @property + def query_params(self) -> MultiDict[Any]: + """Return the query parameters of this connection's ``Scope``. + + Returns: + A normalized dict of query parameters. Multiple values for the same key are returned as a list. + """ + if self._parsed_query is Empty: + if (parsed_query := self._connection_state.parsed_query) is not Empty: + self._parsed_query = parsed_query + else: + self._connection_state.parsed_query = self._parsed_query = parse_query_string( + self.scope.get("query_string", b"") + ) + return MultiDict(self._parsed_query) + + @property + def path_params(self) -> dict[str, Any]: + """Return the ``path_params`` of this connection's ``Scope``. + + Returns: + A string keyed dictionary of path parameter values. + """ + return self.scope["path_params"] + + @property + def cookies(self) -> dict[str, str]: + """Return the ``cookies`` of this connection's ``Scope``. + + Returns: + Returns any cookies stored in the header as a parsed dictionary. + """ + if self._cookies is Empty: + if (cookies := self._connection_state.cookies) is not Empty: + self._cookies = cookies + else: + self._connection_state.cookies = self._cookies = ( + parse_cookie_string(cookie_header) if (cookie_header := self.headers.get("cookie")) else {} + ) + return self._cookies + + @property + def client(self) -> Address | None: + """Return the ``client`` data of this connection's ``Scope``. + + Returns: + A two tuple of the host name and port number. + """ + client = self.scope.get("client") + return Address(*client) if client else None + + @property + def auth(self) -> AuthT: + """Return the ``auth`` data of this connection's ``Scope``. + + Raises: + ImproperlyConfiguredException: If ``auth`` is not set in scope via an ``AuthMiddleware``, raises an exception + + Returns: + A type correlating to the generic variable Auth. + """ + if "auth" not in self.scope: + raise ImproperlyConfiguredException("'auth' is not defined in scope, install an AuthMiddleware to set it") + + return cast("AuthT", self.scope["auth"]) + + @property + def user(self) -> UserT: + """Return the ``user`` data of this connection's ``Scope``. + + Raises: + ImproperlyConfiguredException: If ``user`` is not set in scope via an ``AuthMiddleware``, raises an exception + + Returns: + A type correlating to the generic variable User. + """ + if "user" not in self.scope: + raise ImproperlyConfiguredException("'user' is not defined in scope, install an AuthMiddleware to set it") + + return cast("UserT", self.scope["user"]) + + @property + def session(self) -> dict[str, Any]: + """Return the session for this connection if a session was previously set in the ``Scope`` + + Returns: + A dictionary representing the session value - if existing. + + Raises: + ImproperlyConfiguredException: if session is not set in scope. + """ + if "session" not in self.scope: + raise ImproperlyConfiguredException( + "'session' is not defined in scope, install a SessionMiddleware to set it" + ) + + return cast("dict[str, Any]", self.scope["session"]) + + @property + def logger(self) -> Logger: + """Return the ``Logger`` instance for this connection. + + Returns: + A ``Logger`` instance. + + Raises: + ImproperlyConfiguredException: if ``log_config`` has not been passed to the Litestar constructor. + """ + return self.app.get_logger() + + def set_session(self, value: dict[str, Any] | DataContainerType | EmptyType) -> None: + """Set the session in the connection's ``Scope``. + + If the :class:`SessionMiddleware <.middleware.session.base.SessionMiddleware>` is enabled, the session will be added + to the response as a cookie header. + + Args: + value: Dictionary or pydantic model instance for the session data. + + Returns: + None + """ + self.scope["session"] = value + + def clear_session(self) -> None: + """Remove the session from the connection's ``Scope``. + + If the :class:`Litestar SessionMiddleware <.middleware.session.base.SessionMiddleware>` is enabled, this will cause + the session data to be cleared. + + Returns: + None. + """ + self.scope["session"] = Empty + self._connection_state.session_id = Empty + + def get_session_id(self) -> str | None: + return value_or_default(value=self._connection_state.session_id, default=None) + + def url_for(self, name: str, **path_parameters: Any) -> str: + """Return the url for a given route handler name. + + Args: + name: The ``name`` of the request route handler. + **path_parameters: Values for path parameters in the route + + Raises: + NoRouteMatchFoundException: If route with ``name`` does not exist, path parameters are missing or have a + wrong type. + + Returns: + A string representing the absolute url of the route handler. + """ + litestar_instance = self.scope["app"] + url_path = litestar_instance.route_reverse(name, **path_parameters) + + return make_absolute_url(url_path, self.base_url) + + def url_for_static_asset(self, name: str, file_path: str) -> str: + """Receives a static files handler name, an asset file path and returns resolved absolute url to the asset. + + Args: + name: A static handler unique name. + file_path: a string containing path to an asset. + + Raises: + NoRouteMatchFoundException: If static files handler with ``name`` does not exist. + + Returns: + A string representing absolute url to the asset. + """ + litestar_instance = self.scope["app"] + url_path = litestar_instance.url_for_static_asset(name, file_path) + + return make_absolute_url(url_path, self.base_url) |