genichin
4fa770e771
Release: REL-003_20260512 — CLI user scoping and local DB ops runbook Requirements & Batches: - BAT-003 (REQ-013): portfolio list --user required enforcement with owner filtering - BAT-004 (REQ-014): portfolio create --user required enforcement with owner assignment - BAT-007 (REQ-001/002/003): local DB validation runbook and baseline docs update CLI Enhancements: - Update user.py: add username-based user lookup (get_by_username) - Add UserDetailResponse schema with service token listing - Update list_service_tokens API in user_service.py - Enhance user detail display with service token information - Add service_token_count and service_token_names to user responses Testing: - Add comprehensive test coverage in test_cli.py for user commands - Validate username-based lookup functionality Release: REL-004_20260512 — Portfolio UUID migration and API UUID switch Requirements & Batches: - BAT-006 (REQ-016/017/018): portfolio UUID migration complete - BAT-008 (REQ-019): baseline documentation update for UUID migration Breaking Changes (REL-003): - portfolio list requires --user <username> parameter - portfolio create requires --user <username> parameter Documentation: - Create rel-003 and rel-004 release documents with rollout and rollback plans - Update docs/releases/index.md with new releases - Update README.md with CLI user scoping clarification - Sync all release verification checklists and deployment notes
279 lines
11 KiB
Python
279 lines
11 KiB
Python
"""Formatting and execution helpers for the CLI."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from dataclasses import asdict, is_dataclass
|
|
from typing import Any, Callable, TypeVar
|
|
|
|
import typer
|
|
from pydantic import BaseModel, ValidationError
|
|
from rich.console import Console
|
|
from rich.panel import Panel
|
|
from rich.table import Table
|
|
|
|
from app.core.errors import FeatureNotReadyError
|
|
from app.schemas.auth import AuthTokenResponse
|
|
from app.schemas.portfolio import CommentResponse, PortfolioDetailResponse, PortfolioListItem
|
|
from app.schemas.user import UserResponse
|
|
|
|
T = TypeVar("T")
|
|
|
|
|
|
def format_currency(value: float | None) -> str:
|
|
if value is None:
|
|
return "N/A"
|
|
return f"{value:,.0f}"
|
|
|
|
|
|
def to_plain_data(value: Any) -> Any:
|
|
if isinstance(value, BaseModel):
|
|
return value.model_dump(mode="json")
|
|
if is_dataclass(value):
|
|
return asdict(value)
|
|
if isinstance(value, list):
|
|
return [to_plain_data(item) for item in value]
|
|
if isinstance(value, dict):
|
|
return {key: to_plain_data(item) for key, item in value.items()}
|
|
return value
|
|
|
|
|
|
def print_portfolio_table(console: Console, portfolios: list[PortfolioListItem]) -> None:
|
|
for i, item in enumerate(portfolios):
|
|
if i > 0:
|
|
console.print()
|
|
lines = [
|
|
f"id: {item.portfolio_id}",
|
|
f"broker: {item.broker}",
|
|
f"status: {item.status}",
|
|
f"symbol: {item.symbol}",
|
|
f"cash: {format_currency(item.cash_balance)}",
|
|
f"qty: {item.asset_quantity:g}",
|
|
f"price: {format_currency(item.price)}",
|
|
f"buffer: {format_currency(item.profit_buffer)}",
|
|
f"total: {format_currency(item.total_value)}",
|
|
]
|
|
for line in lines:
|
|
console.print(line)
|
|
|
|
|
|
def print_portfolio_detail(console: Console, detail: PortfolioDetailResponse) -> None:
|
|
portfolio = detail.portfolio
|
|
position_value = float(portfolio.asset_quantity) * float(portfolio.price)
|
|
bot_total = float(portfolio.cash_balance) + position_value
|
|
user_total = bot_total + float(portfolio.profit_buffer)
|
|
panel = Panel.fit(
|
|
"\n".join(
|
|
[
|
|
f"id: {portfolio.portfolio_id}",
|
|
f"broker: {portfolio.broker}",
|
|
f"status: {portfolio.status}",
|
|
f"symbol: {portfolio.symbol}",
|
|
f"active_strategy_id: {portfolio.active_strategy_id or 'N/A'}",
|
|
f"avabile_strategies: {', '.join(portfolio.avabile_strategies) if portfolio.avabile_strategies else 'N/A'}",
|
|
f"cash_balance: {format_currency(portfolio.cash_balance)}",
|
|
f"asset_quantity: {portfolio.asset_quantity:g}",
|
|
f"price: {format_currency(portfolio.price)}",
|
|
f"profit_buffer: {format_currency(portfolio.profit_buffer)}",
|
|
f"position_value: {format_currency(position_value)}",
|
|
f"bot_total: {format_currency(bot_total)}",
|
|
f"user_total: {format_currency(user_total)}",
|
|
f"live_quantity: {detail.live_quantity if detail.live_quantity is not None else 'N/A'}",
|
|
]
|
|
),
|
|
title="Portfolio Status",
|
|
)
|
|
console.print(panel)
|
|
|
|
if detail.strategy_set is not None:
|
|
console.print(
|
|
Panel.fit(
|
|
"\n".join(
|
|
[
|
|
f"id: {detail.strategy_set.id}",
|
|
f"name: {detail.strategy_set.name}",
|
|
f"config: {detail.strategy_set.config}",
|
|
f"nick: {detail.strategy_set.nick}",
|
|
]
|
|
),
|
|
title="Strategy Set",
|
|
)
|
|
)
|
|
|
|
ledger_table = Table(title="Recent Ledger Entries", expand=True)
|
|
ledger_table.add_column("Recorded At", style="cyan", no_wrap=True)
|
|
ledger_table.add_column("Action")
|
|
ledger_table.add_column("Qty", justify="right")
|
|
ledger_table.add_column("Price", justify="right")
|
|
ledger_table.add_column("Cash After", justify="right")
|
|
ledger_table.add_column("Qty After", justify="right")
|
|
ledger_table.add_column("Buffer After", justify="right")
|
|
for entry in detail.recent_entries:
|
|
ledger_table.add_row(
|
|
entry.recorded_at.isoformat(timespec="seconds"),
|
|
entry.action,
|
|
f"{entry.quantity:g}",
|
|
format_currency(entry.price),
|
|
format_currency(entry.cash_balance_after),
|
|
f"{entry.asset_quantity_after:g}",
|
|
format_currency(entry.profit_buffer_after),
|
|
)
|
|
console.print(ledger_table)
|
|
|
|
|
|
def print_user_table(console: Console, users: list[UserResponse]) -> None:
|
|
for i, user in enumerate(users, 1):
|
|
lines = [
|
|
f"username: {user.username}",
|
|
f"id: {user.id}",
|
|
f"email: {user.email or 'N/A'}",
|
|
f"permission: {user.permission}",
|
|
f"is_active: {'true' if user.is_active else 'false'}",
|
|
f"created_at: {user.created_at.isoformat(timespec='seconds') if user.created_at else 'N/A'}",
|
|
f"updated_at: {user.updated_at.isoformat(timespec='seconds') if user.updated_at else 'N/A'}",
|
|
]
|
|
for line in lines:
|
|
console.print(line)
|
|
if i < len(users):
|
|
console.print()
|
|
|
|
|
|
def print_user_detail(console: Console, user: UserResponse) -> None:
|
|
lines = [
|
|
f"id: {user.id}",
|
|
f"username: {user.username}",
|
|
f"email: {user.email or 'N/A'}",
|
|
f"permission: {user.permission}",
|
|
f"is_active: {'true' if user.is_active else 'false'}",
|
|
f"last_login_at: {user.last_login_at.isoformat(timespec='seconds') if user.last_login_at else 'N/A'}",
|
|
f"created_at: {user.created_at.isoformat(timespec='seconds') if user.created_at else 'N/A'}",
|
|
f"updated_at: {user.updated_at.isoformat(timespec='seconds') if user.updated_at else 'N/A'}",
|
|
]
|
|
service_token_count = getattr(user, "service_token_count", None)
|
|
service_token_names = getattr(user, "service_token_names", None)
|
|
if service_token_count is not None and service_token_names is not None:
|
|
rendered_names = ", ".join(service_token_names) if service_token_names else "N/A"
|
|
lines.append(f"service_token_count: {service_token_count}")
|
|
lines.append(f"service_token_names: {rendered_names}")
|
|
for line in lines:
|
|
console.print(line)
|
|
|
|
|
|
def print_service_token_list(console: Console, tokens: list[AuthTokenResponse]) -> None:
|
|
for i, token in enumerate(tokens, 1):
|
|
status = "[red]revoked[/red]" if token.revoked_at else "[green]active[/green]"
|
|
expires = token.expires_at.isoformat(timespec="seconds") if token.expires_at else "never"
|
|
last_used = token.last_used_at.isoformat(timespec="seconds") if token.last_used_at else "-"
|
|
revoked_at = token.revoked_at.isoformat(timespec="seconds") if token.revoked_at else "-"
|
|
created_at = token.created_at.isoformat(timespec="seconds") if token.created_at else "N/A"
|
|
lines = [
|
|
f"[bold cyan]{token.name}[/bold cyan] ({status})",
|
|
f" token_prefix: {token.token_prefix} [dim](full token shown only at issue time)[/dim]",
|
|
f" id : {token.id}",
|
|
f" user_id : {token.user_id}",
|
|
f" perm : {token.permission}",
|
|
f" expires : {expires}",
|
|
f" last_used: {last_used}",
|
|
f" created : {created_at}",
|
|
]
|
|
if token.revoked_at:
|
|
lines.append(f" revoked : {revoked_at}")
|
|
console.print("\n".join(lines))
|
|
if i < len(tokens):
|
|
console.print("[dim]─[/dim]" * 40)
|
|
|
|
|
|
# Keep old name as alias for backward compatibility
|
|
print_service_token_table = print_service_token_list
|
|
|
|
|
|
def print_comment_table(console: Console, comments: list[CommentResponse]) -> None:
|
|
table = Table(title="Portfolio Comments", expand=True)
|
|
table.add_column("ID", style="cyan", no_wrap=True)
|
|
table.add_column("Portfolio ID", no_wrap=True)
|
|
table.add_column("Content")
|
|
table.add_column("Created At", no_wrap=True)
|
|
table.add_column("Updated At", no_wrap=True)
|
|
for comment in comments:
|
|
table.add_row(
|
|
comment.id,
|
|
comment.portfolio_id,
|
|
comment.content,
|
|
comment.created_at.isoformat(timespec="seconds") if comment.created_at else "N/A",
|
|
comment.updated_at.isoformat(timespec="seconds") if comment.updated_at else "N/A",
|
|
)
|
|
console.print(table)
|
|
|
|
|
|
def print_comment_detail(console: Console, comment: CommentResponse) -> None:
|
|
console.print(
|
|
Panel.fit(
|
|
"\n".join(
|
|
[
|
|
f"id: {comment.id}",
|
|
f"portfolio_id: {comment.portfolio_id}",
|
|
f"content: {comment.content}",
|
|
f"created_at: {comment.created_at.isoformat(timespec='seconds') if comment.created_at else 'N/A'}",
|
|
f"updated_at: {comment.updated_at.isoformat(timespec='seconds') if comment.updated_at else 'N/A'}",
|
|
]
|
|
),
|
|
title="Comment",
|
|
)
|
|
)
|
|
|
|
|
|
def print_json(console: Console, payload: Any) -> None:
|
|
console.print_json(data=to_plain_data(payload))
|
|
|
|
|
|
def render_output(console: Console, output_format: str, payload: Any, *, empty_message: str | None = None) -> None:
|
|
if output_format == "json":
|
|
print_json(console, payload)
|
|
return
|
|
|
|
if isinstance(payload, list):
|
|
if not payload and empty_message:
|
|
console.print(empty_message)
|
|
return
|
|
if all(isinstance(item, PortfolioListItem) for item in payload):
|
|
print_portfolio_table(console, payload)
|
|
return
|
|
if all(isinstance(item, UserResponse) for item in payload):
|
|
print_user_table(console, payload)
|
|
return
|
|
if all(isinstance(item, CommentResponse) for item in payload):
|
|
print_comment_table(console, payload)
|
|
return
|
|
if all(isinstance(item, AuthTokenResponse) for item in payload):
|
|
print_service_token_table(console, payload)
|
|
return
|
|
return
|
|
|
|
if isinstance(payload, PortfolioDetailResponse):
|
|
print_portfolio_detail(console, payload)
|
|
return
|
|
|
|
if isinstance(payload, UserResponse):
|
|
print_user_detail(console, payload)
|
|
return
|
|
|
|
if isinstance(payload, CommentResponse):
|
|
print_comment_detail(console, payload)
|
|
return
|
|
|
|
console.print(payload)
|
|
|
|
|
|
def execute_command(console: Console, callback: Callable[[], T]) -> T:
|
|
try:
|
|
return callback()
|
|
except ValidationError as exc:
|
|
console.print(f"Input validation failed: {exc}")
|
|
raise typer.Exit(code=1) from exc
|
|
except FeatureNotReadyError as exc:
|
|
console.print(str(exc))
|
|
raise typer.Exit(code=1) from exc
|
|
except ValueError as exc:
|
|
console.print(str(exc))
|
|
raise typer.Exit(code=1) from exc
|