diff options
Diffstat (limited to 'venv/lib/python3.11/site-packages/litestar/utils/scope')
-rw-r--r-- | venv/lib/python3.11/site-packages/litestar/utils/scope/__init__.py | 62 | ||||
-rw-r--r-- | venv/lib/python3.11/site-packages/litestar/utils/scope/__pycache__/__init__.cpython-311.pyc | bin | 2727 -> 0 bytes | |||
-rw-r--r-- | venv/lib/python3.11/site-packages/litestar/utils/scope/__pycache__/state.cpython-311.pyc | bin | 6672 -> 0 bytes | |||
-rw-r--r-- | venv/lib/python3.11/site-packages/litestar/utils/scope/state.py | 158 |
4 files changed, 0 insertions, 220 deletions
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 deleted file mode 100644 index e5757d3..0000000 --- a/venv/lib/python3.11/site-packages/litestar/utils/scope/__init__.py +++ /dev/null @@ -1,62 +0,0 @@ -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 Binary files differdeleted file mode 100644 index ee218d5..0000000 --- a/venv/lib/python3.11/site-packages/litestar/utils/scope/__pycache__/__init__.cpython-311.pyc +++ /dev/null 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 Binary files differdeleted file mode 100644 index 4615004..0000000 --- a/venv/lib/python3.11/site-packages/litestar/utils/scope/__pycache__/state.cpython-311.pyc +++ /dev/null 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 deleted file mode 100644 index bed4394..0000000 --- a/venv/lib/python3.11/site-packages/litestar/utils/scope/state.py +++ /dev/null @@ -1,158 +0,0 @@ -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] |