summaryrefslogtreecommitdiff
path: root/venv/lib/python3.11/site-packages/litestar/utils/scope
diff options
context:
space:
mode:
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__.py62
-rw-r--r--venv/lib/python3.11/site-packages/litestar/utils/scope/__pycache__/__init__.cpython-311.pycbin2727 -> 0 bytes
-rw-r--r--venv/lib/python3.11/site-packages/litestar/utils/scope/__pycache__/state.cpython-311.pycbin6672 -> 0 bytes
-rw-r--r--venv/lib/python3.11/site-packages/litestar/utils/scope/state.py158
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
deleted file mode 100644
index ee218d5..0000000
--- a/venv/lib/python3.11/site-packages/litestar/utils/scope/__pycache__/__init__.cpython-311.pyc
+++ /dev/null
Binary files 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
deleted file mode 100644
index 4615004..0000000
--- a/venv/lib/python3.11/site-packages/litestar/utils/scope/__pycache__/state.cpython-311.pyc
+++ /dev/null
Binary files 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
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]