75 lines
2.5 KiB
Python
75 lines
2.5 KiB
Python
import os
|
|
import shutil
|
|
import uuid
|
|
import logging
|
|
from datetime import datetime, timezone
|
|
from typing import List
|
|
from .config import Config
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
class LocalSessionStorage:
|
|
"""Per-request session folder storage with TTL cleanup."""
|
|
|
|
def __init__(self, base_dir: str | None = None):
|
|
self.base_dir = base_dir or Config.SESSIONS_DIR
|
|
os.makedirs(self.base_dir, exist_ok=True)
|
|
|
|
@staticmethod
|
|
def _now_utc() -> datetime:
|
|
return datetime.now(timezone.utc)
|
|
|
|
def _session_path(self, session_id: str) -> str:
|
|
# defensive join to prevent path traversal
|
|
candidate = os.path.realpath(os.path.join(self.base_dir, session_id))
|
|
if not candidate.startswith(os.path.realpath(self.base_dir)):
|
|
raise ValueError("Invalid session path")
|
|
return candidate
|
|
|
|
def new_session(self) -> str:
|
|
sid = str(uuid.uuid4())
|
|
path = self._session_path(sid)
|
|
os.makedirs(path, exist_ok=True)
|
|
return sid
|
|
|
|
def save_bytes(self, session_id: str, filename: str, content: bytes) -> str:
|
|
path = self._session_path(session_id)
|
|
os.makedirs(path, exist_ok=True)
|
|
safe_name = os.path.basename(filename)
|
|
full = os.path.join(path, safe_name)
|
|
with open(full, 'wb') as f:
|
|
f.write(content)
|
|
return full
|
|
|
|
def list_files(self, session_id: str) -> List[str]:
|
|
path = self._session_path(session_id)
|
|
if not os.path.isdir(path):
|
|
return []
|
|
return [os.path.join(path, p) for p in os.listdir(path) if os.path.isfile(os.path.join(path, p))]
|
|
|
|
def delete_session(self, session_id: str) -> bool:
|
|
path = self._session_path(session_id)
|
|
if os.path.isdir(path):
|
|
shutil.rmtree(path, ignore_errors=True)
|
|
return True
|
|
return False
|
|
|
|
def cleanup_expired(self) -> int:
|
|
"""Remove sessions older than TTL; returns count removed."""
|
|
removed = 0
|
|
ttl = Config.SESSION_TTL
|
|
now = self._now_utc()
|
|
for sid in os.listdir(self.base_dir):
|
|
spath = self._session_path(sid)
|
|
try:
|
|
mtime = datetime.fromtimestamp(os.path.getmtime(spath), tz=timezone.utc)
|
|
if now - mtime > ttl:
|
|
shutil.rmtree(spath, ignore_errors=True)
|
|
removed += 1
|
|
except Exception as e:
|
|
log.warning(f"cleanup skip {sid}: {e}")
|
|
if removed:
|
|
log.info(f"Session cleanup removed {removed} dirs")
|
|
return removed
|