From 6d7ba58f880be618ade07f8ea080fe8c4bf8a896 Mon Sep 17 00:00:00 2001
From: cyfraeviolae <cyfraeviolae>
Date: Wed, 3 Apr 2024 03:10:44 -0400
Subject: venv

---
 .../litestar/cli/commands/sessions.py              | 58 ++++++++++++++++++++++
 1 file changed, 58 insertions(+)
 create mode 100644 venv/lib/python3.11/site-packages/litestar/cli/commands/sessions.py

(limited to 'venv/lib/python3.11/site-packages/litestar/cli/commands/sessions.py')

diff --git a/venv/lib/python3.11/site-packages/litestar/cli/commands/sessions.py b/venv/lib/python3.11/site-packages/litestar/cli/commands/sessions.py
new file mode 100644
index 0000000..f048dd1
--- /dev/null
+++ b/venv/lib/python3.11/site-packages/litestar/cli/commands/sessions.py
@@ -0,0 +1,58 @@
+from click import argument, group
+from rich.prompt import Confirm
+
+from litestar import Litestar
+from litestar.cli._utils import LitestarCLIException, LitestarGroup, console
+from litestar.middleware import DefineMiddleware
+from litestar.middleware.session import SessionMiddleware
+from litestar.middleware.session.server_side import ServerSideSessionBackend
+from litestar.utils import is_class_and_subclass
+
+__all__ = ("clear_sessions_command", "delete_session_command", "get_session_backend", "sessions_group")
+
+
+def get_session_backend(app: Litestar) -> ServerSideSessionBackend:
+    """Get the session backend used by a ``Litestar`` app."""
+    for middleware in app.middleware:
+        if isinstance(middleware, DefineMiddleware):
+            if not is_class_and_subclass(middleware.middleware, SessionMiddleware):
+                continue
+            backend = middleware.kwargs["backend"]
+            if not isinstance(backend, ServerSideSessionBackend):
+                raise LitestarCLIException("Only server-side backends are supported")
+            return backend
+    raise LitestarCLIException("Session middleware not installed")
+
+
+@group(cls=LitestarGroup, name="sessions")
+def sessions_group() -> None:
+    """Manage server-side sessions."""
+
+
+@sessions_group.command("delete")  # type: ignore[misc]
+@argument("session-id")
+def delete_session_command(session_id: str, app: Litestar) -> None:
+    """Delete a specific session."""
+    import anyio
+
+    backend = get_session_backend(app)
+    store = backend.config.get_store_from_app(app)
+
+    if Confirm.ask(f"Delete session {session_id!r}?"):
+        anyio.run(backend.delete, session_id, store)
+        console.print(f"[green]Deleted session {session_id!r}")
+
+
+@sessions_group.command("clear")  # type: ignore[misc]
+def clear_sessions_command(app: Litestar) -> None:
+    """Delete all sessions."""
+    import anyio
+
+    backend = get_session_backend(app)
+    store = backend.config.get_store_from_app(app)
+    if not hasattr(store, "delete_all"):
+        raise LitestarCLIException(f"{type(store)} does not support clearing all sessions")
+
+    if Confirm.ask("[red]Delete all sessions?"):
+        anyio.run(store.delete_all)  # pyright: ignore
+        console.print("[green]All active sessions deleted")
-- 
cgit v1.2.3