feat(PORTFOLIO-001): add position-level portfolio entries
This commit is contained in:
@@ -2,9 +2,10 @@ from __future__ import annotations
|
||||
|
||||
import re
|
||||
from pathlib import Path
|
||||
from uuid import uuid4
|
||||
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(
|
||||
@@ -63,6 +64,69 @@ class WorkspaceRepository:
|
||||
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"
|
||||
|
||||
|
||||
Reference in New Issue
Block a user