from __future__ import annotations import re from pathlib import Path from uuid import UUID, uuid4 from app.models.portfolio import PortfolioConfig, PortfolioRepository, build_default_portfolio_config from app.models.position import Position 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, *, config: PortfolioConfig | 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") created_config = config or build_default_portfolio_config() self.save_portfolio_config(resolved_workspace_id, created_config) return created_config def create_workspace_id(self, *, config: PortfolioConfig | None = None) -> str: workspace_id = str(uuid4()) self.create_workspace(workspace_id, config=config) 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 add_position(self, workspace_id: str, position: Position) -> None: """Add a position to the workspace portfolio.""" if not self.is_valid_workspace_id(workspace_id): raise ValueError("workspace_id must be a UUID4 string") config = self.load_portfolio_config(workspace_id) # Use object.__setattr__ because positions is in a frozen dataclass object.__setattr__(config, "positions", list(config.positions) + [position]) self.save_portfolio_config(workspace_id, config) def remove_position(self, workspace_id: str, position_id: UUID) -> None: """Remove a position from the workspace portfolio.""" if not self.is_valid_workspace_id(workspace_id): raise ValueError("workspace_id must be a UUID4 string") config = self.load_portfolio_config(workspace_id) updated_positions = [p for p in config.positions if p.id != position_id] object.__setattr__(config, "positions", updated_positions) self.save_portfolio_config(workspace_id, config) def update_position( self, workspace_id: str, position_id: UUID, updates: dict[str, object], ) -> None: """Update a position's fields.""" if not self.is_valid_workspace_id(workspace_id): raise ValueError("workspace_id must be a UUID4 string") config = self.load_portfolio_config(workspace_id) updated_positions = [] for pos in config.positions: if pos.id == position_id: # Create updated position (Position is frozen, so create new instance) update_kwargs: dict[str, object] = {} for key, value in updates.items(): if key in {"id", "created_at"}: continue # Skip immutable fields update_kwargs[key] = value # Use dataclass replace-like pattern pos_dict = pos.to_dict() pos_dict.update(update_kwargs) updated_positions.append(Position.from_dict(pos_dict)) else: updated_positions.append(pos) object.__setattr__(config, "positions", updated_positions) self.save_portfolio_config(workspace_id, config) def get_position(self, workspace_id: str, position_id: UUID) -> Position | None: """Get a specific position by ID.""" if not self.is_valid_workspace_id(workspace_id): raise ValueError("workspace_id must be a UUID4 string") config = self.load_portfolio_config(workspace_id) for pos in config.positions: if pos.id == position_id: return pos return None def list_positions(self, workspace_id: str) -> list[Position]: """List all positions in the workspace portfolio.""" if not self.is_valid_workspace_id(workspace_id): raise ValueError("workspace_id must be a UUID4 string") config = self.load_portfolio_config(workspace_id) return list(config.positions) 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