Spaces:
Sleeping
Sleeping
File size: 1,628 Bytes
acd245a |
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 |
"""Session manager for AiderService."""
import uuid
from typing import Dict, Optional
from app.services.aider_service import AiderService
class SessionManager:
"""Session manager for AiderService."""
def __init__(self) -> None:
"""Initialize the session manager."""
self.sessions: Dict[str, AiderService] = {}
async def create_session(self, repo_url: Optional[str] = None) -> str:
"""Create a new session and return its ID.
Args:
repo_url: Optional URL to a git repository to clone. If provided,
the repository will be cloned instead of initializing an empty repo.
Returns:
The session ID as a string.
"""
session_id = str(uuid.uuid4())
aider_service = AiderService()
await aider_service.initialize(repo_url=repo_url)
await aider_service.start_aider()
self.sessions[session_id] = aider_service
return session_id
def get_session(self, session_id: str) -> Optional[AiderService]:
"""Get an existing session by ID."""
return self.sessions.get(session_id)
async def cleanup_session(self, session_id: str) -> None:
"""Clean up a session and remove it from the manager."""
if session_id in self.sessions:
await self.sessions[session_id].cleanup()
del self.sessions[session_id]
async def cleanup_all_sessions(self) -> None:
"""Clean up all sessions."""
for session_id in list(self.sessions.keys()):
await self.cleanup_session(session_id)
session_manager = SessionManager()
|