diff options
Diffstat (limited to 'venv/lib/python3.11/site-packages/litestar/middleware/session')
8 files changed, 0 insertions, 742 deletions
diff --git a/venv/lib/python3.11/site-packages/litestar/middleware/session/__init__.py b/venv/lib/python3.11/site-packages/litestar/middleware/session/__init__.py deleted file mode 100644 index 1ca9c17..0000000 --- a/venv/lib/python3.11/site-packages/litestar/middleware/session/__init__.py +++ /dev/null @@ -1,3 +0,0 @@ -from .base import SessionMiddleware - -__all__ = ("SessionMiddleware",) diff --git a/venv/lib/python3.11/site-packages/litestar/middleware/session/__pycache__/__init__.cpython-311.pyc b/venv/lib/python3.11/site-packages/litestar/middleware/session/__pycache__/__init__.cpython-311.pyc Binary files differdeleted file mode 100644 index 8748ce3..0000000 --- a/venv/lib/python3.11/site-packages/litestar/middleware/session/__pycache__/__init__.cpython-311.pyc +++ /dev/null diff --git a/venv/lib/python3.11/site-packages/litestar/middleware/session/__pycache__/base.cpython-311.pyc b/venv/lib/python3.11/site-packages/litestar/middleware/session/__pycache__/base.cpython-311.pyc Binary files differdeleted file mode 100644 index 68a8b9c..0000000 --- a/venv/lib/python3.11/site-packages/litestar/middleware/session/__pycache__/base.cpython-311.pyc +++ /dev/null diff --git a/venv/lib/python3.11/site-packages/litestar/middleware/session/__pycache__/client_side.cpython-311.pyc b/venv/lib/python3.11/site-packages/litestar/middleware/session/__pycache__/client_side.cpython-311.pyc Binary files differdeleted file mode 100644 index 692f54c..0000000 --- a/venv/lib/python3.11/site-packages/litestar/middleware/session/__pycache__/client_side.cpython-311.pyc +++ /dev/null diff --git a/venv/lib/python3.11/site-packages/litestar/middleware/session/__pycache__/server_side.cpython-311.pyc b/venv/lib/python3.11/site-packages/litestar/middleware/session/__pycache__/server_side.cpython-311.pyc Binary files differdeleted file mode 100644 index bd2373c..0000000 --- a/venv/lib/python3.11/site-packages/litestar/middleware/session/__pycache__/server_side.cpython-311.pyc +++ /dev/null diff --git a/venv/lib/python3.11/site-packages/litestar/middleware/session/base.py b/venv/lib/python3.11/site-packages/litestar/middleware/session/base.py deleted file mode 100644 index a823848..0000000 --- a/venv/lib/python3.11/site-packages/litestar/middleware/session/base.py +++ /dev/null @@ -1,256 +0,0 @@ -from __future__ import annotations - -from abc import ABC, abstractmethod -from typing import ( - TYPE_CHECKING, - Any, - Awaitable, - Callable, - Generic, - Literal, - TypeVar, - cast, -) - -from litestar.connection import ASGIConnection -from litestar.enums import ScopeType -from litestar.middleware.base import AbstractMiddleware, DefineMiddleware -from litestar.serialization import decode_json, encode_json -from litestar.utils import get_serializer_from_scope - -__all__ = ("BaseBackendConfig", "BaseSessionBackend", "SessionMiddleware") - - -if TYPE_CHECKING: - from litestar.types import ASGIApp, Message, Receive, Scope, Scopes, ScopeSession, Send - -ONE_DAY_IN_SECONDS = 60 * 60 * 24 - -ConfigT = TypeVar("ConfigT", bound="BaseBackendConfig") -BaseSessionBackendT = TypeVar("BaseSessionBackendT", bound="BaseSessionBackend") - - -class BaseBackendConfig(ABC, Generic[BaseSessionBackendT]): # pyright: ignore - """Configuration for Session middleware backends.""" - - _backend_class: type[BaseSessionBackendT] # pyright: ignore - - key: str - """Key to use for the cookie inside the header, e.g. ``session=<data>`` where ``session`` is the cookie key and - ``<data>`` is the session data. - - Notes: - - If a session cookie exceeds 4KB in size it is split. In this case the key will be of the format - ``session-{segment number}``. - - """ - max_age: int - """Maximal age of the cookie before its invalidated.""" - scopes: Scopes = {ScopeType.HTTP, ScopeType.WEBSOCKET} - """Scopes for the middleware - options are ``http`` and ``websocket`` with the default being both""" - path: str - """Path fragment that must exist in the request url for the cookie to be valid. - - Defaults to ``'/'``. - """ - domain: str | None - """Domain for which the cookie is valid.""" - secure: bool - """Https is required for the cookie.""" - httponly: bool - """Forbids javascript to access the cookie via 'Document.cookie'.""" - samesite: Literal["lax", "strict", "none"] - """Controls whether or not a cookie is sent with cross-site requests. - - Defaults to ``lax``. - """ - exclude: str | list[str] | None - """A pattern or list of patterns to skip in the session middleware.""" - exclude_opt_key: str - """An identifier to use on routes to disable the session middleware for a particular route.""" - - @property - def middleware(self) -> DefineMiddleware: - """Use this property to insert the config into a middleware list on one of the application layers. - - Examples: - .. code-block:: python - - from os import urandom - - from litestar import Litestar, Request, get - from litestar.middleware.sessions.cookie_backend import CookieBackendConfig - - session_config = CookieBackendConfig(secret=urandom(16)) - - - @get("/") - def my_handler(request: Request) -> None: ... - - - app = Litestar(route_handlers=[my_handler], middleware=[session_config.middleware]) - - - Returns: - An instance of DefineMiddleware including ``self`` as the config kwarg value. - """ - return DefineMiddleware(SessionMiddleware, backend=self._backend_class(config=self)) - - -class BaseSessionBackend(ABC, Generic[ConfigT]): - """Abstract session backend defining the interface between a storage mechanism and the application - :class:`SessionMiddleware`. - - This serves as the base class for all client- and server-side backends - """ - - __slots__ = ("config",) - - def __init__(self, config: ConfigT) -> None: - """Initialize ``BaseSessionBackend`` - - Args: - config: A instance of a subclass of ``BaseBackendConfig`` - """ - self.config = config - - @staticmethod - def serialize_data(data: ScopeSession, scope: Scope | None = None) -> bytes: - """Serialize data into bytes for storage in the backend. - - Args: - data: Session data of the current scope. - scope: A scope, if applicable, from which to extract a serializer. - - Notes: - - The serializer will be extracted from ``scope`` or fall back to - :func:`default_serializer <.serialization.default_serializer>` - - Returns: - ``data`` serialized as bytes. - """ - serializer = get_serializer_from_scope(scope) if scope else None - return encode_json(data, serializer) - - @staticmethod - def deserialize_data(data: Any) -> dict[str, Any]: - """Deserialize data into a dictionary for use in the application scope. - - Args: - data: Data to be deserialized - - Returns: - Deserialized data as a dictionary - """ - return cast("dict[str, Any]", decode_json(value=data)) - - @abstractmethod - def get_session_id(self, connection: ASGIConnection) -> str | None: - """Try to fetch session id from connection ScopeState. If one does not exist, generate one. - - Args: - connection: Originating ASGIConnection containing the scope - - Returns: - Session id str or None if the concept of a session id does not apply. - """ - - @abstractmethod - async def store_in_message(self, scope_session: ScopeSession, message: Message, connection: ASGIConnection) -> None: - """Store the necessary information in the outgoing ``Message`` - - Args: - scope_session: Current session to store - message: Outgoing send-message - connection: Originating ASGIConnection containing the scope - - Returns: - None - """ - - @abstractmethod - async def load_from_connection(self, connection: ASGIConnection) -> dict[str, Any]: - """Load session data from a connection and return it as a dictionary to be used in the current application - scope. - - Args: - connection: An ASGIConnection instance - - Returns: - The session data - - Notes: - - This should not modify the connection's scope. The data returned by this - method will be stored in the application scope by the middleware - - """ - - -class SessionMiddleware(AbstractMiddleware, Generic[BaseSessionBackendT]): - """Litestar session middleware for storing session data.""" - - def __init__(self, app: ASGIApp, backend: BaseSessionBackendT) -> None: - """Initialize ``SessionMiddleware`` - - Args: - app: An ASGI application - backend: A :class:`BaseSessionBackend` instance used to store and retrieve session data - """ - - super().__init__( - app=app, - exclude=backend.config.exclude, - exclude_opt_key=backend.config.exclude_opt_key, - scopes=backend.config.scopes, - ) - self.backend = backend - - def create_send_wrapper(self, connection: ASGIConnection) -> Callable[[Message], Awaitable[None]]: - """Create a wrapper for the ASGI send function, which handles setting the cookies on the outgoing response. - - Args: - connection: ASGIConnection - - Returns: - None - """ - - async def wrapped_send(message: Message) -> None: - """Wrap the ``send`` function. - - Declared in local scope to make use of closure values. - - Args: - message: An ASGI message. - - Returns: - None - """ - if message["type"] != "http.response.start": - await connection.send(message) - return - - scope_session = connection.scope.get("session") - - await self.backend.store_in_message(scope_session, message, connection) - await connection.send(message) - - return wrapped_send - - async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: - """ASGI-callable. - - Args: - scope: The ASGI connection scope. - receive: The ASGI receive function. - send: The ASGI send function. - - Returns: - None - """ - - connection = ASGIConnection[Any, Any, Any, Any](scope, receive=receive, send=send) - scope["session"] = await self.backend.load_from_connection(connection) - connection._connection_state.session_id = self.backend.get_session_id(connection) # pyright: ignore [reportGeneralTypeIssues] - - await self.app(scope, receive, self.create_send_wrapper(connection)) diff --git a/venv/lib/python3.11/site-packages/litestar/middleware/session/client_side.py b/venv/lib/python3.11/site-packages/litestar/middleware/session/client_side.py deleted file mode 100644 index f709410..0000000 --- a/venv/lib/python3.11/site-packages/litestar/middleware/session/client_side.py +++ /dev/null @@ -1,264 +0,0 @@ -from __future__ import annotations - -import binascii -import contextlib -import re -import time -from base64 import b64decode, b64encode -from dataclasses import dataclass, field -from os import urandom -from typing import TYPE_CHECKING, Any, Literal - -from litestar.datastructures import MutableScopeHeaders -from litestar.datastructures.cookie import Cookie -from litestar.enums import ScopeType -from litestar.exceptions import ( - ImproperlyConfiguredException, - MissingDependencyException, -) -from litestar.serialization import decode_json, encode_json -from litestar.types import Empty, Scopes -from litestar.utils.dataclass import extract_dataclass_items - -from .base import ONE_DAY_IN_SECONDS, BaseBackendConfig, BaseSessionBackend - -__all__ = ("ClientSideSessionBackend", "CookieBackendConfig") - - -try: - from cryptography.exceptions import InvalidTag - from cryptography.hazmat.primitives.ciphers.aead import AESGCM -except ImportError as e: - raise MissingDependencyException("cryptography") from e - -if TYPE_CHECKING: - from litestar.connection import ASGIConnection - from litestar.types import Message, Scope, ScopeSession - -NONCE_SIZE = 12 -CHUNK_SIZE = 4096 - 64 -AAD = b"additional_authenticated_data=" - - -class ClientSideSessionBackend(BaseSessionBackend["CookieBackendConfig"]): - """Cookie backend for SessionMiddleware.""" - - __slots__ = ("aesgcm", "cookie_re") - - def __init__(self, config: CookieBackendConfig) -> None: - """Initialize ``ClientSideSessionBackend``. - - Args: - config: SessionCookieConfig instance. - """ - super().__init__(config) - self.aesgcm = AESGCM(config.secret) - self.cookie_re = re.compile(rf"{self.config.key}(?:-\d+)?") - - def dump_data(self, data: Any, scope: Scope | None = None) -> list[bytes]: - """Given serializable data, including pydantic models and numpy types, dump it into a bytes string, encrypt, - encode and split it into chunks of the desirable size. - - Args: - data: Data to serialize, encrypt, encode and chunk. - scope: The ASGI connection scope. - - Notes: - - The returned list is composed of a chunks of a single base64 encoded - string that is encrypted using AES-CGM. - - Returns: - List of encoded bytes string of a maximum length equal to the ``CHUNK_SIZE`` constant. - """ - serialized = self.serialize_data(data, scope) - associated_data = encode_json({"expires_at": round(time.time()) + self.config.max_age}) - nonce = urandom(NONCE_SIZE) - encrypted = self.aesgcm.encrypt(nonce, serialized, associated_data=associated_data) - encoded = b64encode(nonce + encrypted + AAD + associated_data) - return [encoded[i : i + CHUNK_SIZE] for i in range(0, len(encoded), CHUNK_SIZE)] - - def load_data(self, data: list[bytes]) -> dict[str, Any]: - """Given a list of strings, decodes them into the session object. - - Args: - data: A list of strings derived from the request's session cookie(s). - - Returns: - A deserialized session value. - """ - decoded = b64decode(b"".join(data)) - nonce = decoded[:NONCE_SIZE] - aad_starts_from = decoded.find(AAD) - associated_data = decoded[aad_starts_from:].replace(AAD, b"") if aad_starts_from != -1 else None - if associated_data and decode_json(value=associated_data)["expires_at"] > round(time.time()): - encrypted_session = decoded[NONCE_SIZE:aad_starts_from] - decrypted = self.aesgcm.decrypt(nonce, encrypted_session, associated_data=associated_data) - return self.deserialize_data(decrypted) - return {} - - def get_cookie_keys(self, connection: ASGIConnection) -> list[str]: - """Return a list of cookie-keys from the connection if they match the session-cookie pattern. - - Args: - connection: An ASGIConnection instance - - Returns: - A list of session-cookie keys - """ - return sorted(key for key in connection.cookies if self.cookie_re.fullmatch(key)) - - def _create_session_cookies(self, data: list[bytes], cookie_params: dict[str, Any] | None = None) -> list[Cookie]: - """Create a list of cookies containing the session data. - If the data is split into multiple cookies, the key will be of the format ``session-{segment number}``, - however if only one cookie is needed, the key will be ``session``. - """ - if cookie_params is None: - cookie_params = dict( - extract_dataclass_items( - self.config, - exclude_none=True, - include={f for f in Cookie.__dict__ if f not in ("key", "secret")}, - ) - ) - - if len(data) == 1: - return [ - Cookie( - value=data[0].decode("utf-8"), - key=self.config.key, - **cookie_params, - ) - ] - - return [ - Cookie( - value=datum.decode("utf-8"), - key=f"{self.config.key}-{i}", - **cookie_params, - ) - for i, datum in enumerate(data) - ] - - async def store_in_message(self, scope_session: ScopeSession, message: Message, connection: ASGIConnection) -> None: - """Store data from ``scope_session`` in ``Message`` in the form of cookies. If the contents of ``scope_session`` - are too large to fit a single cookie, it will be split across several cookies, following the naming scheme of - ``<cookie key>-<n>``. If the session is empty or shrinks, cookies will be cleared by setting their value to - ``"null"`` - - Args: - scope_session: Current session to store - message: Outgoing send-message - connection: Originating ASGIConnection containing the scope - - Returns: - None - """ - - scope = connection.scope - headers = MutableScopeHeaders.from_message(message) - cookie_keys = self.get_cookie_keys(connection) - - if scope_session and scope_session is not Empty: - data = self.dump_data(scope_session, scope=scope) - cookie_params = dict( - extract_dataclass_items( - self.config, - exclude_none=True, - include={f for f in Cookie.__dict__ if f not in ("key", "secret")}, - ) - ) - for cookie in self._create_session_cookies(data, cookie_params): - headers.add("Set-Cookie", cookie.to_header(header="")) - # Cookies with the same key overwrite the earlier cookie with that key. To expire earlier session - # cookies, first check how many session cookies will not be overwritten in this upcoming response. - # If leftover cookies are greater than or equal to 1, that means older session cookies have to be - # expired and their names are in cookie_keys. - cookies_to_clear = cookie_keys[len(data) :] if len(cookie_keys) - len(data) > 0 else [] - else: - cookies_to_clear = cookie_keys - - for cookie_key in cookies_to_clear: - cookie_params = dict( - extract_dataclass_items( - self.config, - exclude_none=True, - include={f for f in Cookie.__dict__ if f not in ("key", "secret", "max_age")}, - ) - ) - headers.add( - "Set-Cookie", - Cookie(value="null", key=cookie_key, expires=0, **cookie_params).to_header(header=""), - ) - - async def load_from_connection(self, connection: ASGIConnection) -> dict[str, Any]: - """Load session data from a connection's session-cookies and return it as a dictionary. - - Args: - connection: Originating ASGIConnection - - Returns: - The session data - """ - if cookie_keys := self.get_cookie_keys(connection): - data = [connection.cookies[key].encode("utf-8") for key in cookie_keys] - # If these exceptions occur, the session must remain empty so do nothing. - with contextlib.suppress(InvalidTag, binascii.Error): - return self.load_data(data) - return {} - - def get_session_id(self, connection: ASGIConnection) -> str | None: - return None - - -@dataclass -class CookieBackendConfig(BaseBackendConfig[ClientSideSessionBackend]): # pyright: ignore - """Configuration for [SessionMiddleware] middleware.""" - - _backend_class = ClientSideSessionBackend - - secret: bytes - """A secret key to use for generating an encryption key. - - Must have a length of 16 (128 bits), 24 (192 bits) or 32 (256 bits) characters. - """ - key: str = field(default="session") - """Key to use for the cookie inside the header, e.g. ``session=<data>`` where ``session`` is the cookie key and - ``<data>`` is the session data. - - Notes: - - If a session cookie exceeds 4KB in size it is split. In this case the key will be of the format - ``session-{segment number}``. - - """ - max_age: int = field(default=ONE_DAY_IN_SECONDS * 14) - """Maximal age of the cookie before its invalidated.""" - scopes: Scopes = field(default_factory=lambda: {ScopeType.HTTP, ScopeType.WEBSOCKET}) - """Scopes for the middleware - options are ``http`` and ``websocket`` with the default being both""" - path: str = field(default="/") - """Path fragment that must exist in the request url for the cookie to be valid. - - Defaults to ``'/'``. - """ - domain: str | None = field(default=None) - """Domain for which the cookie is valid.""" - secure: bool = field(default=False) - """Https is required for the cookie.""" - httponly: bool = field(default=True) - """Forbids javascript to access the cookie via 'Document.cookie'.""" - samesite: Literal["lax", "strict", "none"] = field(default="lax") - """Controls whether or not a cookie is sent with cross-site requests. - - Defaults to ``lax``. - """ - exclude: str | list[str] | None = field(default=None) - """A pattern or list of patterns to skip in the session middleware.""" - exclude_opt_key: str = field(default="skip_session") - """An identifier to use on routes to disable the session middleware for a particular route.""" - - def __post_init__(self) -> None: - if len(self.key) < 1 or len(self.key) > 256: - raise ImproperlyConfiguredException("key must be a string with a length between 1-256") - if self.max_age < 1: - raise ImproperlyConfiguredException("max_age must be greater than 0") - if len(self.secret) not in {16, 24, 32}: - raise ImproperlyConfiguredException("secret length must be 16 (128 bit), 24 (192 bit) or 32 (256 bit)") diff --git a/venv/lib/python3.11/site-packages/litestar/middleware/session/server_side.py b/venv/lib/python3.11/site-packages/litestar/middleware/session/server_side.py deleted file mode 100644 index 91708ac..0000000 --- a/venv/lib/python3.11/site-packages/litestar/middleware/session/server_side.py +++ /dev/null @@ -1,219 +0,0 @@ -from __future__ import annotations - -import secrets -from dataclasses import dataclass, field -from typing import TYPE_CHECKING, Any, Literal - -from litestar.datastructures import Cookie, MutableScopeHeaders -from litestar.enums import ScopeType -from litestar.exceptions import ImproperlyConfiguredException -from litestar.middleware.session.base import ONE_DAY_IN_SECONDS, BaseBackendConfig, BaseSessionBackend -from litestar.types import Empty, Message, Scopes, ScopeSession -from litestar.utils.dataclass import extract_dataclass_items - -__all__ = ("ServerSideSessionBackend", "ServerSideSessionConfig") - - -if TYPE_CHECKING: - from litestar import Litestar - from litestar.connection import ASGIConnection - from litestar.stores.base import Store - - -class ServerSideSessionBackend(BaseSessionBackend["ServerSideSessionConfig"]): - """Base class for server-side backends. - - Implements :class:`BaseSessionBackend` and defines and interface which subclasses can - implement to facilitate the storage of session data. - """ - - def __init__(self, config: ServerSideSessionConfig) -> None: - """Initialize ``ServerSideSessionBackend`` - - Args: - config: A subclass of ``ServerSideSessionConfig`` - """ - super().__init__(config=config) - - async def get(self, session_id: str, store: Store) -> bytes | None: - """Retrieve data associated with ``session_id``. - - Args: - session_id: The session-ID - store: Store to retrieve the session data from - - Returns: - The session data, if existing, otherwise ``None``. - """ - max_age = int(self.config.max_age) if self.config.max_age is not None else None - return await store.get(session_id, renew_for=max_age if self.config.renew_on_access else None) - - async def set(self, session_id: str, data: bytes, store: Store) -> None: - """Store ``data`` under the ``session_id`` for later retrieval. - - If there is already data associated with ``session_id``, replace - it with ``data`` and reset its expiry time - - Args: - session_id: The session-ID - data: Serialized session data - store: Store to save the session data in - - Returns: - None - """ - expires_in = int(self.config.max_age) if self.config.max_age is not None else None - await store.set(session_id, data, expires_in=expires_in) - - async def delete(self, session_id: str, store: Store) -> None: - """Delete the data associated with ``session_id``. Fails silently if no such session-ID exists. - - Args: - session_id: The session-ID - store: Store to delete the session data from - - Returns: - None - """ - await store.delete(session_id) - - def get_session_id(self, connection: ASGIConnection) -> str: - """Try to fetch session id from the connection. If one does not exist, generate one. - - If a session ID already exists in the cookies, it is returned. - If there is no ID in the cookies but one in the connection state, then the session exists but has not yet - been returned to the user. - Otherwise, a new session must be created. - - Args: - connection: Originating ASGIConnection containing the scope - Returns: - Session id str or None if the concept of a session id does not apply. - """ - session_id = connection.cookies.get(self.config.key) - if not session_id or session_id == "null": - session_id = connection.get_session_id() - if not session_id: - session_id = self.generate_session_id() - return session_id - - def generate_session_id(self) -> str: - """Generate a new session-ID, with - n=:attr:`session_id_bytes <ServerSideSessionConfig.session_id_bytes>` random bytes. - - Returns: - A session-ID - """ - return secrets.token_hex(self.config.session_id_bytes) - - async def store_in_message(self, scope_session: ScopeSession, message: Message, connection: ASGIConnection) -> None: - """Store the necessary information in the outgoing ``Message`` by setting a cookie containing the session-ID. - - If the session is empty, a null-cookie will be set. Otherwise, the serialised - data will be stored using :meth:`set <ServerSideSessionBackend.set>`, under the current session-id. If no session-ID - exists, a new ID will be generated using :meth:`generate_session_id <ServerSideSessionBackend.generate_session_id>`. - - Args: - scope_session: Current session to store - message: Outgoing send-message - connection: Originating ASGIConnection containing the scope - - Returns: - None - """ - scope = connection.scope - store = self.config.get_store_from_app(scope["app"]) - headers = MutableScopeHeaders.from_message(message) - session_id = self.get_session_id(connection) - - cookie_params = dict(extract_dataclass_items(self.config, exclude_none=True, include=Cookie.__dict__.keys())) - - if scope_session is Empty: - await self.delete(session_id, store=store) - headers.add( - "Set-Cookie", - Cookie(value="null", key=self.config.key, expires=0, **cookie_params).to_header(header=""), - ) - else: - serialised_data = self.serialize_data(scope_session, scope) - await self.set(session_id=session_id, data=serialised_data, store=store) - headers.add( - "Set-Cookie", Cookie(value=session_id, key=self.config.key, **cookie_params).to_header(header="") - ) - - async def load_from_connection(self, connection: ASGIConnection) -> dict[str, Any]: - """Load session data from a connection and return it as a dictionary to be used in the current application - scope. - - The session-ID will be gathered from a cookie with the key set in - :attr:`BaseBackendConfig.key`. If a cookie is found, its value will be used as the session-ID and data associated - with this ID will be loaded using :meth:`get <ServerSideSessionBackend.get>`. - If no cookie was found or no data was loaded from the store, this will return an - empty dictionary. - - Args: - connection: An ASGIConnection instance - - Returns: - The current session data - """ - if session_id := connection.cookies.get(self.config.key): - store = self.config.get_store_from_app(connection.scope["app"]) - data = await self.get(session_id, store=store) - if data is not None: - return self.deserialize_data(data) - return {} - - -@dataclass -class ServerSideSessionConfig(BaseBackendConfig[ServerSideSessionBackend]): # pyright: ignore - """Base configuration for server side backends.""" - - _backend_class = ServerSideSessionBackend - - session_id_bytes: int = field(default=32) - """Number of bytes used to generate a random session-ID.""" - renew_on_access: bool = field(default=False) - """Renew expiry times of sessions when they're being accessed""" - key: str = field(default="session") - """Key to use for the cookie inside the header, e.g. ``session=<data>`` where ``session`` is the cookie key and - ``<data>`` is the session data. - - Notes: - - If a session cookie exceeds 4KB in size it is split. In this case the key will be of the format - ``session-{segment number}``. - - """ - max_age: int = field(default=ONE_DAY_IN_SECONDS * 14) - """Maximal age of the cookie before its invalidated.""" - scopes: Scopes = field(default_factory=lambda: {ScopeType.HTTP, ScopeType.WEBSOCKET}) - """Scopes for the middleware - options are ``http`` and ``websocket`` with the default being both""" - path: str = field(default="/") - """Path fragment that must exist in the request url for the cookie to be valid. - - Defaults to ``'/'``. - """ - domain: str | None = field(default=None) - """Domain for which the cookie is valid.""" - secure: bool = field(default=False) - """Https is required for the cookie.""" - httponly: bool = field(default=True) - """Forbids javascript to access the cookie via 'Document.cookie'.""" - samesite: Literal["lax", "strict", "none"] = field(default="lax") - """Controls whether or not a cookie is sent with cross-site requests. Defaults to ``lax``.""" - exclude: str | list[str] | None = field(default=None) - """A pattern or list of patterns to skip in the session middleware.""" - exclude_opt_key: str = field(default="skip_session") - """An identifier to use on routes to disable the session middleware for a particular route.""" - store: str = "sessions" - """Name of the :class:`Store <.stores.base.Store>` to use""" - - def __post_init__(self) -> None: - if len(self.key) < 1 or len(self.key) > 256: - raise ImproperlyConfiguredException("key must be a string with a length between 1-256") - if self.max_age < 1: - raise ImproperlyConfiguredException("max_age must be greater than 0") - - def get_store_from_app(self, app: Litestar) -> Store: - """Get the store defined in :attr:`store` from an :class:`Litestar <.app.Litestar>` instance""" - return app.stores.get(self.store) |