Spaces:
Sleeping
Sleeping
File size: 3,762 Bytes
b336134 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 | """
Session manager — lightweight in-memory session metadata.
In production, swap this dict for Redis. Each entry holds only
metadata (not the dataframe itself). The actual data lives on
disk as a Parquet file under DATA_DIR/{session_id}.parquet.
"""
from __future__ import annotations
import time
import threading
from dataclasses import dataclass, field
from typing import Optional
from config import DATA_DIR, SESSION_TTL_MINUTES
@dataclass
class SessionMeta:
session_id: str
file_name: str
file_size_bytes: int
columns: list[dict[str, str]] # [{"name": ..., "dtype": ...}]
row_count: int
status: str = "active"
current_version: int = 0
created_at: float = field(default_factory=time.time)
last_active: float = field(default_factory=time.time)
def touch(self) -> None:
self.last_active = time.time()
class SessionManager:
"""Thread-safe session store."""
def __init__(self) -> None:
self._sessions: dict[str, SessionMeta] = {}
self._lock = threading.Lock()
def create(
self,
session_id: str,
file_name: str,
file_size_bytes: int,
columns: list[dict[str, str]],
row_count: int,
) -> SessionMeta:
meta = SessionMeta(
session_id=session_id,
file_name=file_name,
file_size_bytes=file_size_bytes,
columns=columns,
row_count=row_count,
)
with self._lock:
self._sessions[session_id] = meta
return meta
def get(self, session_id: str) -> Optional[SessionMeta]:
with self._lock:
return self._sessions.get(session_id)
def get_filepath(self, session_id: str) -> str:
"""Get the absolute filepath of the current version of the Parquet file."""
meta = self.get(session_id)
if meta:
version = getattr(meta, "current_version", 0)
if version > 0:
v_path = os.path.join(DATA_DIR, f"{session_id}_v{version}.parquet")
if os.path.exists(v_path):
return v_path
# Version 0 or fallback: check v0 path first
v0_path = os.path.join(DATA_DIR, f"{session_id}_v0.parquet")
if os.path.exists(v0_path):
return v0_path
return os.path.join(DATA_DIR, f"{session_id}.parquet")
def touch(self, session_id: str) -> None:
meta = self.get(session_id)
if meta:
meta.touch()
def remove(self, session_id: str) -> None:
with self._lock:
self._sessions.pop(session_id, None)
def list_active(self) -> list[SessionMeta]:
"""Return sessions that haven't expired."""
now = time.time()
cutoff = now - SESSION_TTL_MINUTES * 60
with self._lock:
return [
m for m in self._sessions.values()
if m.last_active > cutoff
]
def cleanup_expired(self) -> int:
"""Remove expired sessions and their Parquet files. Returns count removed."""
import os
import glob
now = time.time()
cutoff = now - SESSION_TTL_MINUTES * 60
removed = 0
with self._lock:
expired = [sid for sid, m in self._sessions.items() if m.last_active <= cutoff]
for sid in expired:
del self._sessions[sid]
pattern = os.path.join(DATA_DIR, f"{sid}*.parquet")
for pq in glob.glob(pattern):
try:
os.remove(pq)
except Exception:
pass
removed += 1
return removed
# Module-level singleton
session_manager = SessionManager() |