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/datastructures/state.py | 313 +++++++++++++++++++++ 1 file changed, 313 insertions(+) create mode 100644 venv/lib/python3.11/site-packages/litestar/datastructures/state.py (limited to 'venv/lib/python3.11/site-packages/litestar/datastructures/state.py') diff --git a/venv/lib/python3.11/site-packages/litestar/datastructures/state.py b/venv/lib/python3.11/site-packages/litestar/datastructures/state.py new file mode 100644 index 0000000..71980e0 --- /dev/null +++ b/venv/lib/python3.11/site-packages/litestar/datastructures/state.py @@ -0,0 +1,313 @@ +from __future__ import annotations + +from copy import copy, deepcopy +from threading import RLock +from typing import TYPE_CHECKING, Any, Callable, Generator, Iterable, Iterator, Mapping, MutableMapping + +if TYPE_CHECKING: + from typing_extensions import Self + +__all__ = ("ImmutableState", "State") + + +class ImmutableState(Mapping[str, Any]): + """An object meant to store arbitrary state. + + It can be accessed using dot notation while exposing dict like functionalities. + """ + + __slots__ = ( + "_state", + "_deep_copy", + ) + + _state: dict[str, Any] + + def __init__( + self, state: ImmutableState | Mapping[str, Any] | Iterable[tuple[str, Any]], deep_copy: bool = True + ) -> None: + """Initialize an ``ImmutableState`` instance. + + Args: + state: An object to initialize the state from. Can be a dict, an instance of :class:`ImmutableState`, or a tuple + of key value paris. + deep_copy: Whether to 'deepcopy' the passed in state. + + Examples: + .. code-block:: python + + from litestar.datastructures import ImmutableState + + state_dict = {"first": 1, "second": 2, "third": 3, "fourth": 4} + state = ImmutableState(state_dict) + + # state implements the Mapping type: + assert len(state) == 3 + assert "first" in state + assert not "fourth" in state + assert state["first"] == 1 + assert [(k, v) for k, v in state.items()] == [("first", 1), ("second", 2), ("third", 3)] + + # state implements __bool__ + assert state # state is true when it has values. + assert not State() # state is empty when it has no values. + + # it has a 'dict' method to retrieve a shallow copy of the underlying dict + inner_dict = state.dict() + assert inner_dict == state_dict + + # you can also retrieve a mutable State by calling 'mutable_copy' + mutable_state = state.mutable_copy() + del state["first"] + assert "first" not in state + + """ + if isinstance(state, ImmutableState): + state = state._state + + if not isinstance(state, dict) and isinstance(state, Iterable): + state = dict(state) + + super().__setattr__("_deep_copy", deep_copy) + super().__setattr__("_state", deepcopy(state) if deep_copy else state) + + def __bool__(self) -> bool: + """Return a boolean indicating whether the wrapped dict instance has values.""" + return bool(self._state) + + def __getitem__(self, key: str) -> Any: + """Get the value for the corresponding key from the wrapped state object using subscription notation. + + Args: + key: Key to access. + + Raises: + KeyError + + Returns: + A value from the wrapped state instance. + """ + return self._state[key] + + def __iter__(self) -> Iterator[str]: + """Return an iterator iterating the wrapped state dict. + + Returns: + An iterator of strings + """ + return iter(self._state) + + def __len__(self) -> int: + """Return length of the wrapped state dict. + + Returns: + An integer + """ + return len(self._state) + + def __getattr__(self, key: str) -> Any: + """Get the value for the corresponding key from the wrapped state object using attribute notation. + + Args: + key: Key to retrieve + + Raises: + AttributeError: if the given attribute is not set. + + Returns: + The retrieved value + """ + try: + return self._state[key] + except KeyError as e: + raise AttributeError from e + + def __copy__(self) -> Self: + """Return a shallow copy of the given state object. + + Customizes how the builtin "copy" function will work. + """ + return self.__class__(self._state, deep_copy=self._deep_copy) # pyright: ignore + + def mutable_copy(self) -> State: + """Return a mutable copy of the state object. + + Returns: + A ``State`` + """ + return State(self._state, deep_copy=self._deep_copy) + + def dict(self) -> dict[str, Any]: + """Return a shallow copy of the wrapped dict. + + Returns: + A dict + """ + return copy(self._state) + + @classmethod + def __get_validators__( + cls, + ) -> Generator[Callable[[ImmutableState | dict[str, Any] | Iterable[tuple[str, Any]]], ImmutableState], None, None]: # type: ignore[valid-type] + """Pydantic compatible method to allow custom parsing of state instances in a SignatureModel.""" + yield cls.validate + + @classmethod + def validate(cls, value: ImmutableState | dict[str, Any] | Iterable[tuple[str, Any]]) -> Self: # type: ignore[valid-type] + """Parse a value and instantiate state inside a SignatureModel. This allows us to use custom subclasses of + state, as well as allows users to decide whether state is mutable or immutable. + + Args: + value: The value from which to initialize the state instance. + + Returns: + An ImmutableState instance + """ + deep_copy = value._deep_copy if isinstance(value, ImmutableState) else False + return cls(value, deep_copy=deep_copy) + + +class State(ImmutableState, MutableMapping[str, Any]): + """An object meant to store arbitrary state. + + It can be accessed using dot notation while exposing dict like functionalities. + """ + + __slots__ = ("_lock",) + + _lock: RLock + + def __init__( + self, + state: ImmutableState | Mapping[str, Any] | Iterable[tuple[str, Any]] | None = None, + deep_copy: bool = False, + ) -> None: + """Initialize a ``State`` instance with an optional value. + + Args: + state: An object to initialize the state from. Can be a dict, an instance of 'ImmutableState', or a tuple of key value paris. + deep_copy: Whether to 'deepcopy' the passed in state. + + .. code-block:: python + :caption: Examples + + from litestar.datastructures import State + + state_dict = {"first": 1, "second": 2, "third": 3, "fourth": 4} + state = State(state_dict) + + # state can be accessed using '.' notation + assert state.fourth == 4 + del state.fourth + + # state implements the Mapping type: + assert len(state) == 3 + assert "first" in state + assert not "fourth" in state + assert state["first"] == 1 + assert [(k, v) for k, v in state.items()] == [("first", 1), ("second", 2), ("third", 3)] + + state["fourth"] = 4 + assert "fourth" in state + del state["fourth"] + + # state implements __bool__ + assert state # state is true when it has values. + assert not State() # state is empty when it has no values. + + # it has shallow copy + copied_state = state.copy() + del copied_state.first + assert state.first + + # it has a 'dict' method to retrieve a shallow copy of the underlying dict + inner_dict = state.dict() + assert inner_dict == state_dict + + # you can get an immutable copy of the state by calling 'immutable_immutable_copy' + immutable_copy = state.immutable_copy() + del immutable_copy.first # raises AttributeError + + """ + + super().__init__(state if state is not None else {}, deep_copy=deep_copy) + super().__setattr__("_lock", RLock()) + + def __delitem__(self, key: str) -> None: + """Delete the value from the key from the wrapped state object using subscription notation. + + Args: + key: Key to delete + + Raises: + KeyError: if the given attribute is not set. + + Returns: + None + """ + + with self._lock: + del self._state[key] + + def __setitem__(self, key: str, value: Any) -> None: + """Set an item in the state using subscription notation. + + Args: + key: Key to set. + value: Value to set. + + Returns: + None + """ + + with self._lock: + self._state[key] = value + + def __setattr__(self, key: str, value: Any) -> None: + """Set an item in the state using attribute notation. + + Args: + key: Key to set. + value: Value to set. + + Returns: + None + """ + + with self._lock: + self._state[key] = value + + def __delattr__(self, key: str) -> None: + """Delete the value from the key from the wrapped state object using attribute notation. + + Args: + key: Key to delete + + Raises: + AttributeError: if the given attribute is not set. + + Returns: + None + """ + + try: + with self._lock: + del self._state[key] + except KeyError as e: + raise AttributeError from e + + def copy(self) -> Self: + """Return a shallow copy of the state object. + + Returns: + A ``State`` + """ + return self.__class__(self.dict(), deep_copy=self._deep_copy) # pyright: ignore + + def immutable_copy(self) -> ImmutableState: + """Return a shallow copy of the state object, setting it to be frozen. + + Returns: + A ``State`` + """ + return ImmutableState(self, deep_copy=self._deep_copy) -- cgit v1.2.3