73 lines
2.6 KiB
Python
73 lines
2.6 KiB
Python
from __future__ import annotations
|
|
|
|
import re
|
|
from pathlib import Path
|
|
from uuid import uuid4
|
|
|
|
from app.models.portfolio import PortfolioConfig, PortfolioRepository
|
|
|
|
WORKSPACE_COOKIE = "workspace_id"
|
|
_WORKSPACE_ID_RE = re.compile(
|
|
r"^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$",
|
|
re.IGNORECASE,
|
|
)
|
|
|
|
|
|
class WorkspaceRepository:
|
|
"""Persist workspace-scoped portfolio configuration on disk."""
|
|
|
|
def __init__(self, base_path: Path | str = Path("data/workspaces")) -> None:
|
|
self.base_path = Path(base_path)
|
|
self.base_path.mkdir(parents=True, exist_ok=True)
|
|
|
|
def is_valid_workspace_id(self, workspace_id: str) -> bool:
|
|
return bool(_WORKSPACE_ID_RE.match(workspace_id))
|
|
|
|
def workspace_exists(self, workspace_id: str) -> bool:
|
|
if not self.is_valid_workspace_id(workspace_id):
|
|
return False
|
|
portfolio_path = self._portfolio_path(workspace_id)
|
|
if not portfolio_path.exists():
|
|
return False
|
|
try:
|
|
PortfolioRepository(portfolio_path).load()
|
|
except (ValueError, TypeError, FileNotFoundError):
|
|
return False
|
|
return True
|
|
|
|
def create_workspace(self, workspace_id: str | None = None) -> PortfolioConfig:
|
|
resolved_workspace_id = workspace_id or str(uuid4())
|
|
if not self.is_valid_workspace_id(resolved_workspace_id):
|
|
raise ValueError("workspace_id must be a UUID4 string")
|
|
config = PortfolioConfig()
|
|
self.save_portfolio_config(resolved_workspace_id, config)
|
|
return config
|
|
|
|
def create_workspace_id(self) -> str:
|
|
workspace_id = str(uuid4())
|
|
self.create_workspace(workspace_id)
|
|
return workspace_id
|
|
|
|
def load_portfolio_config(self, workspace_id: str) -> PortfolioConfig:
|
|
if not self.workspace_exists(workspace_id):
|
|
raise FileNotFoundError(f"Unknown workspace: {workspace_id}")
|
|
return PortfolioRepository(self._portfolio_path(workspace_id)).load()
|
|
|
|
def save_portfolio_config(self, workspace_id: str, config: PortfolioConfig) -> None:
|
|
if not self.is_valid_workspace_id(workspace_id):
|
|
raise ValueError("workspace_id must be a UUID4 string")
|
|
PortfolioRepository(self._portfolio_path(workspace_id)).save(config)
|
|
|
|
def _portfolio_path(self, workspace_id: str) -> Path:
|
|
return self.base_path / workspace_id / "portfolio_config.json"
|
|
|
|
|
|
_workspace_repo: WorkspaceRepository | None = None
|
|
|
|
|
|
def get_workspace_repository() -> WorkspaceRepository:
|
|
global _workspace_repo
|
|
if _workspace_repo is None:
|
|
_workspace_repo = WorkspaceRepository()
|
|
return _workspace_repo
|