latte
d957120eb3
All checks were successful
Enterprise AI Code Review / ai-review (pull_request) Successful in 38s
165 lines
4.1 KiB
Python
165 lines
4.1 KiB
Python
"""Session management for CLI."""
|
|
|
|
import json
|
|
import secrets
|
|
from dataclasses import asdict, dataclass
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
@dataclass
|
|
class SessionData:
|
|
"""Local session data."""
|
|
|
|
session_id: str
|
|
name: str
|
|
created_at: str
|
|
last_active: str
|
|
message_count: int = 0
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
"""Convert to dictionary.
|
|
|
|
Returns:
|
|
dict: Session data as dictionary
|
|
"""
|
|
return asdict(self)
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict[str, Any]) -> "SessionData":
|
|
"""Create from dictionary.
|
|
|
|
Args:
|
|
data: Dictionary data
|
|
|
|
Returns:
|
|
SessionData: Session instance
|
|
"""
|
|
return cls(**data)
|
|
|
|
|
|
class SessionManager:
|
|
"""Manages local CLI sessions."""
|
|
|
|
def __init__(self, sessions_file: Path):
|
|
"""Initialize session manager.
|
|
|
|
Args:
|
|
sessions_file: Path to sessions file
|
|
"""
|
|
self.sessions_file = sessions_file
|
|
self.sessions: dict[str, SessionData] = {}
|
|
self._load()
|
|
|
|
def _load(self) -> None:
|
|
"""Load sessions from file."""
|
|
if self.sessions_file.exists():
|
|
try:
|
|
with open(self.sessions_file, "r") as f:
|
|
data = json.load(f)
|
|
|
|
self.sessions = {
|
|
name: SessionData.from_dict(session_data) for name, session_data in data.items()
|
|
}
|
|
except (json.JSONDecodeError, IOError) as e:
|
|
print(f"Warning: Could not load sessions: {e}")
|
|
self.sessions = {}
|
|
|
|
def _save(self) -> None:
|
|
"""Save sessions to file."""
|
|
self.sessions_file.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
data = {name: session.to_dict() for name, session in self.sessions.items()}
|
|
|
|
with open(self.sessions_file, "w") as f:
|
|
json.dump(data, f, indent=2)
|
|
|
|
def create_session(self, name: str = "default") -> SessionData:
|
|
"""Create or get a session.
|
|
|
|
Args:
|
|
name: Session name
|
|
|
|
Returns:
|
|
SessionData: Created or existing session
|
|
"""
|
|
if name in self.sessions:
|
|
return self.sessions[name]
|
|
|
|
# Generate unique session ID
|
|
session_id = f"cli_{name}_{secrets.token_hex(8)}"
|
|
now = datetime.utcnow().isoformat()
|
|
|
|
session = SessionData(
|
|
session_id=session_id,
|
|
name=name,
|
|
created_at=now,
|
|
last_active=now,
|
|
message_count=0,
|
|
)
|
|
|
|
self.sessions[name] = session
|
|
self._save()
|
|
|
|
return session
|
|
|
|
def get_session(self, name: str) -> SessionData | None:
|
|
"""Get a session by name.
|
|
|
|
Args:
|
|
name: Session name
|
|
|
|
Returns:
|
|
SessionData | None: Session or None if not found
|
|
"""
|
|
return self.sessions.get(name)
|
|
|
|
def get_or_create_session(self, name: str = "default") -> SessionData:
|
|
"""Get or create a session.
|
|
|
|
Args:
|
|
name: Session name
|
|
|
|
Returns:
|
|
SessionData: Session
|
|
"""
|
|
session = self.get_session(name)
|
|
if session:
|
|
return session
|
|
return self.create_session(name)
|
|
|
|
def update_last_active(self, name: str) -> None:
|
|
"""Update session's last active time.
|
|
|
|
Args:
|
|
name: Session name
|
|
"""
|
|
if name in self.sessions:
|
|
self.sessions[name].last_active = datetime.utcnow().isoformat()
|
|
self.sessions[name].message_count += 1
|
|
self._save()
|
|
|
|
def list_sessions(self) -> list[SessionData]:
|
|
"""List all sessions.
|
|
|
|
Returns:
|
|
list[SessionData]: All sessions
|
|
"""
|
|
return sorted(self.sessions.values(), key=lambda s: s.last_active, reverse=True)
|
|
|
|
def delete_session(self, name: str) -> bool:
|
|
"""Delete a session.
|
|
|
|
Args:
|
|
name: Session name
|
|
|
|
Returns:
|
|
bool: True if deleted, False if not found
|
|
"""
|
|
if name in self.sessions:
|
|
del self.sessions[name]
|
|
self._save()
|
|
return True
|
|
return False
|