File size: 4,302 Bytes
b010f1b |
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 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 |
"""
Session Manager - In-memory session storage
Stores site data, layouts, and chat history per session
"""
import uuid
from typing import Dict, Optional, Any, List
from dataclasses import dataclass, field
from datetime import datetime
import threading
@dataclass
class Session:
"""Session data container"""
id: str
created_at: datetime
boundary: Optional[Dict] = None
boundary_coords: Optional[List] = None
metadata: Dict = field(default_factory=dict)
layouts: List[Dict] = field(default_factory=list)
chat_history: List[Dict] = field(default_factory=list)
def to_dict(self) -> Dict:
return {
"id": self.id,
"created_at": self.created_at.isoformat(),
"has_boundary": self.boundary is not None,
"num_layouts": len(self.layouts),
"num_messages": len(self.chat_history)
}
class SessionManager:
"""
In-memory session storage manager
Thread-safe session creation and retrieval.
Sessions are stored in a dictionary with UUID keys.
"""
def __init__(self, max_sessions: int = 1000):
self._sessions: Dict[str, Session] = {}
self._lock = threading.Lock()
self._max_sessions = max_sessions
def create_session(self) -> Session:
"""Create new session with UUID"""
session_id = str(uuid.uuid4())
session = Session(
id=session_id,
created_at=datetime.now()
)
with self._lock:
# Cleanup if too many sessions
if len(self._sessions) >= self._max_sessions:
self._cleanup_oldest()
self._sessions[session_id] = session
return session
def get_session(self, session_id: str) -> Optional[Session]:
"""Get session by ID"""
return self._sessions.get(session_id)
def update_session(self, session_id: str, **kwargs) -> Optional[Session]:
"""Update session data"""
session = self._sessions.get(session_id)
if session:
for key, value in kwargs.items():
if hasattr(session, key):
setattr(session, key, value)
return session
def set_boundary(self, session_id: str, boundary: Dict, coords: List, metadata: Dict) -> bool:
"""Set boundary data for session"""
session = self._sessions.get(session_id)
if session:
session.boundary = boundary
session.boundary_coords = coords
session.metadata = metadata
return True
return False
def set_layouts(self, session_id: str, layouts: List[Dict]) -> bool:
"""Set generated layouts for session"""
session = self._sessions.get(session_id)
if session:
session.layouts = layouts
return True
return False
def add_chat_message(self, session_id: str, role: str, content: str, model: str = None) -> bool:
"""Add chat message to history"""
session = self._sessions.get(session_id)
if session:
session.chat_history.append({
"role": role,
"content": content,
"model": model,
"timestamp": datetime.now().isoformat()
})
return True
return False
def delete_session(self, session_id: str) -> bool:
"""Delete a session"""
with self._lock:
if session_id in self._sessions:
del self._sessions[session_id]
return True
return False
def _cleanup_oldest(self):
"""Remove oldest sessions when limit reached"""
if not self._sessions:
return
# Sort by created_at and remove oldest 10%
sorted_sessions = sorted(
self._sessions.items(),
key=lambda x: x[1].created_at
)
remove_count = max(1, len(sorted_sessions) // 10)
for session_id, _ in sorted_sessions[:remove_count]:
del self._sessions[session_id]
@property
def session_count(self) -> int:
return len(self._sessions)
# Global session manager instance
session_manager = SessionManager()
|