summaryrefslogtreecommitdiff
path: root/venv/lib/python3.11/site-packages/litestar/utils/scope/state.py
diff options
context:
space:
mode:
Diffstat (limited to 'venv/lib/python3.11/site-packages/litestar/utils/scope/state.py')
-rw-r--r--venv/lib/python3.11/site-packages/litestar/utils/scope/state.py158
1 files changed, 0 insertions, 158 deletions
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]