Spaces:
Sleeping
Sleeping
| """Session management: maps browser cookies to Session objects.""" | |
| from __future__ import annotations | |
| import uuid | |
| from dataclasses import dataclass, field | |
| from typing import Any | |
| import duckdb | |
| class Session: | |
| id: str | |
| duckdb_con: duckdb.DuckDBPyConnection | |
| current_table: str | None = None | |
| current_ref: Any = None # DataFrameRef | |
| history: Any = None # History | |
| metadata: dict = field(default_factory=dict) | |
| class SessionStore: | |
| """In-memory session store. In production, use Redis or DB-backed store.""" | |
| def __init__(self) -> None: | |
| self._sessions: dict[str, Session] = {} | |
| def create(self, duckdb_con: duckdb.DuckDBPyConnection) -> Session: | |
| from visual_notebook.domain.history import History | |
| session_id = uuid.uuid4().hex[:12] | |
| session = Session( | |
| id=session_id, | |
| duckdb_con=duckdb_con, | |
| history=History(), | |
| ) | |
| self._sessions[session_id] = session | |
| return session | |
| def get(self, session_id: str) -> Session | None: | |
| return self._sessions.get(session_id) | |
| def get_or_create(self, session_id: str | None, duckdb_con: duckdb.DuckDBPyConnection) -> Session: | |
| if session_id and (session := self.get(session_id)): | |
| return session | |
| return self.create(duckdb_con) | |
| def remove(self, session_id: str) -> None: | |
| self._sessions.pop(session_id, None) | |