"""Strategy Set CLI commands.""" from __future__ import annotations import typer from app.cli.core.utils import execute_command, render_output from app.services.strategy_set_service import StrategySetService strategy_app = typer.Typer(help="Strategy set management commands.") def _with_service(ctx: typer.Context, callback): app_context = ctx.obj repository = app_context.repository_factory(app_context.settings) service = StrategySetService(repository) try: result = execute_command(app_context.console, callback=lambda: callback(service, app_context)) return result finally: if callable(getattr(type(repository), "close", None)): repository.close() def _print_strategy_set_table(strategy_sets): """Print strategy sets as a simple table.""" if not strategy_sets: print("No strategy sets found.") return # Print header print(f"{'ID':<40} {'Name':<20} {'Nick':<20} {'Created At':<26}") print("-" * 106) # Print rows for ss in strategy_sets: created_at = ss.created_at.isoformat() if ss.created_at else "N/A" print(f"{ss.id:<40} {ss.name:<20} {ss.nick:<20} {created_at:<26}") @strategy_app.command("list") def list_strategy_sets(ctx: typer.Context) -> None: """List all available strategy sets.""" def cmd(service: StrategySetService, app_context): strategy_sets = service.list() _print_strategy_set_table(strategy_sets) _with_service(ctx, cmd) @strategy_app.command("create") def create_strategy_set( ctx: typer.Context, name: str = typer.Option(..., help="Strategy name (e.g., 'target_value')."), config: str = typer.Option(..., help="Strategy config as JSON string."), ) -> None: """Create a new strategy set (or retrieve existing if (name, config) pair already exists).""" def cmd(service: StrategySetService, app_context): from app.schemas.portfolio import CreateStrategySetRequest payload = CreateStrategySetRequest(name=name, config=config) strategy_set = service.create(payload) print(f"Strategy set created/retrieved:") print(f" ID: {strategy_set.id}") print(f" Name: {strategy_set.name}") print(f" Nick: {strategy_set.nick}") print(f" Created At: {strategy_set.created_at.isoformat() if strategy_set.created_at else 'N/A'}") _with_service(ctx, cmd) @strategy_app.command("update") def update_strategy_set( ctx: typer.Context, strategy_set_id: str = typer.Argument(..., help="Strategy set ID to update."), name: str | None = typer.Option(None, help="New strategy name (optional)."), config: str | None = typer.Option(None, help="New strategy config as JSON string (optional)."), ) -> None: """Update an existing strategy set's name and/or config.""" def cmd(service: StrategySetService, app_context): from app.schemas.portfolio import UpdateStrategySetRequest payload = UpdateStrategySetRequest(name=name, config=config) strategy_set = service.update(strategy_set_id, payload) print(f"Strategy set updated:") print(f" ID: {strategy_set.id}") print(f" Name: {strategy_set.name}") print(f" Nick: {strategy_set.nick}") print(f" Created At: {strategy_set.created_at.isoformat() if strategy_set.created_at else 'N/A'}") _with_service(ctx, cmd) @strategy_app.command("delete") def delete_strategy_set( ctx: typer.Context, strategy_set_id: str = typer.Argument(..., help="Strategy set ID to delete."), force: bool = typer.Option(False, "--force", "-f", help="Skip confirmation."), ) -> None: """Delete a strategy set. Returns an error if the strategy set is in use by any portfolio. """ def cmd(service: StrategySetService, app_context): if not force: confirmed = typer.confirm(f"Delete strategy set {strategy_set_id}?") if not confirmed: print("Cancelled.") return try: service.delete(strategy_set_id) print(f"Strategy set {strategy_set_id} deleted.") except ValueError as e: print(f"Error: {e}") raise typer.Exit(code=1) _with_service(ctx, cmd)