From 6d7ba58f880be618ade07f8ea080fe8c4bf8a896 Mon Sep 17 00:00:00 2001 From: cyfraeviolae Date: Wed, 3 Apr 2024 03:10:44 -0400 Subject: venv --- .../site-packages/litestar/utils/scope/__init__.py | 62 ++++++++ .../scope/__pycache__/__init__.cpython-311.pyc | Bin 0 -> 2727 bytes .../utils/scope/__pycache__/state.cpython-311.pyc | Bin 0 -> 6672 bytes .../site-packages/litestar/utils/scope/state.py | 158 +++++++++++++++++++++ 4 files changed, 220 insertions(+) create mode 100644 venv/lib/python3.11/site-packages/litestar/utils/scope/__init__.py create mode 100644 venv/lib/python3.11/site-packages/litestar/utils/scope/__pycache__/__init__.cpython-311.pyc create mode 100644 venv/lib/python3.11/site-packages/litestar/utils/scope/__pycache__/state.cpython-311.pyc create mode 100644 venv/lib/python3.11/site-packages/litestar/utils/scope/state.py (limited to 'venv/lib/python3.11/site-packages/litestar/utils/scope') diff --git a/venv/lib/python3.11/site-packages/litestar/utils/scope/__init__.py b/venv/lib/python3.11/site-packages/litestar/utils/scope/__init__.py new file mode 100644 index 0000000..e5757d3 --- /dev/null +++ b/venv/lib/python3.11/site-packages/litestar/utils/scope/__init__.py @@ -0,0 +1,62 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING, Any + +from litestar.serialization import get_serializer +from litestar.utils.deprecation import warn_deprecation +from litestar.utils.scope.state import delete_litestar_scope_state as _delete_litestar_scope_state +from litestar.utils.scope.state import get_litestar_scope_state as _get_litestar_scope_state +from litestar.utils.scope.state import set_litestar_scope_state as _set_litestar_scope_state + +if TYPE_CHECKING: + from litestar.types import Scope, Serializer + +__all__ = ("get_serializer_from_scope",) + + +def get_serializer_from_scope(scope: Scope) -> Serializer: + """Return a serializer given a scope object. + + Args: + scope: The ASGI connection scope. + + Returns: + A serializer function + """ + route_handler = scope["route_handler"] + app = scope["app"] + + if hasattr(route_handler, "resolve_type_encoders"): + type_encoders = route_handler.resolve_type_encoders() + else: + type_encoders = app.type_encoders or {} + + if response_class := ( + route_handler.resolve_response_class() # pyright: ignore + if hasattr(route_handler, "resolve_response_class") + else app.response_class + ): + type_encoders = {**type_encoders, **(response_class.type_encoders or {})} + + return get_serializer(type_encoders) + + +_deprecated_names = { + "get_litestar_scope_state": _get_litestar_scope_state, + "set_litestar_scope_state": _set_litestar_scope_state, + "delete_litestar_scope_state": _delete_litestar_scope_state, +} + + +def __getattr__(name: str) -> Any: + if name in _deprecated_names: + warn_deprecation( + deprecated_name=f"litestar.utils.scope.{name}", + version="2.4", + kind="import", + removal_in="3.0", + info=f"'litestar.utils.scope.{name}' is deprecated. The Litestar scope state is private and should not be " + f"used. Plugin authors should maintain their own scope state namespace.", + ) + return globals()["_deprecated_names"][name] + raise AttributeError(f"module {__name__!r} has no attribute {name!r}") # pragma: no cover diff --git a/venv/lib/python3.11/site-packages/litestar/utils/scope/__pycache__/__init__.cpython-311.pyc b/venv/lib/python3.11/site-packages/litestar/utils/scope/__pycache__/__init__.cpython-311.pyc new file mode 100644 index 0000000..ee218d5 Binary files /dev/null and b/venv/lib/python3.11/site-packages/litestar/utils/scope/__pycache__/__init__.cpython-311.pyc differ diff --git a/venv/lib/python3.11/site-packages/litestar/utils/scope/__pycache__/state.cpython-311.pyc b/venv/lib/python3.11/site-packages/litestar/utils/scope/__pycache__/state.cpython-311.pyc new file mode 100644 index 0000000..4615004 Binary files /dev/null and b/venv/lib/python3.11/site-packages/litestar/utils/scope/__pycache__/state.cpython-311.pyc differ diff --git a/venv/lib/python3.11/site-packages/litestar/utils/scope/state.py b/venv/lib/python3.11/site-packages/litestar/utils/scope/state.py new file mode 100644 index 0000000..bed4394 --- /dev/null +++ b/venv/lib/python3.11/site-packages/litestar/utils/scope/state.py @@ -0,0 +1,158 @@ +from __future__ import annotations + +from dataclasses import dataclass +from typing import TYPE_CHECKING, Any, Final + +from litestar.types import Empty, EmptyType +from litestar.utils.empty import value_or_default + +if TYPE_CHECKING: + from typing_extensions import Self + + from litestar.datastructures import URL, Accept, Headers + from litestar.types.asgi_types import Scope + +CONNECTION_STATE_KEY: Final = "_ls_connection_state" + + +@dataclass +class ScopeState: + """An object for storing connection state. + + This is an internal API, and subject to change without notice. + + All types are a union with `EmptyType` and are seeded with the `Empty` value. + """ + + __slots__ = ( + "accept", + "base_url", + "body", + "content_type", + "cookies", + "csrf_token", + "dependency_cache", + "do_cache", + "form", + "headers", + "is_cached", + "json", + "log_context", + "msgpack", + "parsed_query", + "response_compressed", + "session_id", + "url", + "_compat_ns", + ) + + def __init__(self) -> None: + self.accept = Empty + self.base_url = Empty + self.body = Empty + self.content_type = Empty + self.cookies = Empty + self.csrf_token = Empty + self.dependency_cache = Empty + self.do_cache = Empty + self.form = Empty + self.headers = Empty + self.is_cached = Empty + self.json = Empty + self.log_context: dict[str, Any] = {} + self.msgpack = Empty + self.parsed_query = Empty + self.response_compressed = Empty + self.session_id = Empty + self.url = Empty + self._compat_ns: dict[str, Any] = {} + + accept: Accept | EmptyType + base_url: URL | EmptyType + body: bytes | EmptyType + content_type: tuple[str, dict[str, str]] | EmptyType + cookies: dict[str, str] | EmptyType + csrf_token: str | EmptyType + dependency_cache: dict[str, Any] | EmptyType + do_cache: bool | EmptyType + form: dict[str, str | list[str]] | EmptyType + headers: Headers | EmptyType + is_cached: bool | EmptyType + json: Any | EmptyType + log_context: dict[str, Any] + msgpack: Any | EmptyType + parsed_query: tuple[tuple[str, str], ...] | EmptyType + response_compressed: bool | EmptyType + session_id: str | None | EmptyType + url: URL | EmptyType + _compat_ns: dict[str, Any] + + @classmethod + def from_scope(cls, scope: Scope) -> Self: + """Create a new `ConnectionState` object from a scope. + + Object is cached in the scope's state under the `SCOPE_STATE_NAMESPACE` key. + + Args: + scope: The ASGI connection scope. + + Returns: + A `ConnectionState` object. + """ + base_scope_state = scope.setdefault("state", {}) + if (state := base_scope_state.get(CONNECTION_STATE_KEY)) is None: + state = base_scope_state[CONNECTION_STATE_KEY] = cls() + return state + + +def get_litestar_scope_state(scope: Scope, key: str, default: Any = None, pop: bool = False) -> Any: + """Get an internal value from connection scope state. + + Args: + scope: The connection scope. + key: Key to get from internal namespace in scope state. + default: Default value to return. + pop: Boolean flag dictating whether the value should be deleted from the state. + + Returns: + Value mapped to ``key`` in internal connection scope namespace. + """ + scope_state = ScopeState.from_scope(scope) + try: + val = value_or_default(getattr(scope_state, key), default) + if pop: + setattr(scope_state, key, Empty) + return val + except AttributeError: + if pop: + return scope_state._compat_ns.pop(key, default) + return scope_state._compat_ns.get(key, default) + + +def set_litestar_scope_state(scope: Scope, key: str, value: Any) -> None: + """Set an internal value in connection scope state. + + Args: + scope: The connection scope. + key: Key to set under internal namespace in scope state. + value: Value for key. + """ + scope_state = ScopeState.from_scope(scope) + if hasattr(scope_state, key): + setattr(scope_state, key, value) + else: + scope_state._compat_ns[key] = value + + +def delete_litestar_scope_state(scope: Scope, key: str) -> None: + """Delete an internal value from connection scope state. + + Args: + scope: The connection scope. + key: Key to set under internal namespace in scope state. + """ + scope_state = ScopeState.from_scope(scope) + if hasattr(scope_state, key): + setattr(scope_state, key, Empty) + else: + del scope_state._compat_ns[key] -- cgit v1.2.3