142 lines
5.9 KiB
Python
142 lines
5.9 KiB
Python
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
|