chore: initial project scaffold and gitignore
This commit is contained in:
@@ -0,0 +1 @@
|
||||
"""CLI package for PortfolioHub."""
|
||||
@@ -0,0 +1 @@
|
||||
"""CLI command groups."""
|
||||
@@ -0,0 +1,104 @@
|
||||
"""Health-related CLI commands."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import typer
|
||||
|
||||
from app.cli.core.utils import print_json
|
||||
from app.db.migrations import bootstrap_or_upgrade_to_head
|
||||
from app.repositories.factory import build_portfolio_db_config
|
||||
from app.repositories.postgres import PostgresPortfolioRepository, create_database_if_missing, database_exists
|
||||
|
||||
|
||||
def initialize_database(*, settings, maintenance_db: str = "postgres") -> dict[str, object]:
|
||||
config = build_portfolio_db_config(settings)
|
||||
existed_before = database_exists(config, maintenance_db=maintenance_db)
|
||||
created = False
|
||||
if not existed_before:
|
||||
created = create_database_if_missing(config, maintenance_db=maintenance_db)
|
||||
|
||||
bootstrap_or_upgrade_to_head(settings)
|
||||
|
||||
return {
|
||||
"database": config.dbname,
|
||||
"host": config.host,
|
||||
"port": config.port,
|
||||
"created": created,
|
||||
"schema_initialized": True,
|
||||
"migration_applied": "head",
|
||||
}
|
||||
|
||||
|
||||
def register_health_commands(app: typer.Typer) -> None:
|
||||
@app.command("init-db")
|
||||
def init_db(
|
||||
ctx: typer.Context,
|
||||
maintenance_db: str = typer.Option("postgres", help="Database to connect to for CREATE DATABASE."),
|
||||
) -> None:
|
||||
app_context = ctx.obj
|
||||
try:
|
||||
result = initialize_database(settings=app_context.settings, maintenance_db=maintenance_db)
|
||||
except Exception as exc:
|
||||
app_context.console.print(f"init-db failed: {exc}")
|
||||
raise typer.Exit(code=1) from exc
|
||||
else:
|
||||
if app_context.output_format == "json":
|
||||
print_json(app_context.console, result)
|
||||
else:
|
||||
app_context.console.print(f"database: {result['database']}")
|
||||
app_context.console.print(f"host: {result['host']}")
|
||||
app_context.console.print(f"port: {result['port']}")
|
||||
app_context.console.print(f"created: {result['created']}")
|
||||
app_context.console.print("schema_initialized: True")
|
||||
app_context.console.print("migration_applied: head")
|
||||
|
||||
@app.command("migrate-db")
|
||||
def migrate_db(ctx: typer.Context) -> None:
|
||||
app_context = ctx.obj
|
||||
try:
|
||||
bootstrap_or_upgrade_to_head(app_context.settings)
|
||||
result = {
|
||||
"database": app_context.settings.active_db_name,
|
||||
"host": app_context.settings.active_db_host,
|
||||
"port": app_context.settings.active_db_port,
|
||||
"migration_applied": "head",
|
||||
}
|
||||
except Exception as exc:
|
||||
app_context.console.print(f"migrate-db failed: {exc}")
|
||||
raise typer.Exit(code=1) from exc
|
||||
else:
|
||||
if app_context.output_format == "json":
|
||||
print_json(app_context.console, result)
|
||||
else:
|
||||
app_context.console.print(f"database: {result['database']}")
|
||||
app_context.console.print(f"host: {result['host']}")
|
||||
app_context.console.print(f"port: {result['port']}")
|
||||
app_context.console.print("migration_applied: head")
|
||||
|
||||
@app.command("health-check")
|
||||
def health_check(ctx: typer.Context) -> None:
|
||||
app_context = ctx.obj
|
||||
repository = app_context.repository_factory(app_context.settings)
|
||||
try:
|
||||
health = {
|
||||
"app_name": app_context.settings.app_name,
|
||||
"environment": app_context.settings.environment,
|
||||
"db_enabled": app_context.settings.db_enabled,
|
||||
"active_db_host": app_context.settings.active_db_host,
|
||||
"active_db_port": app_context.settings.active_db_port,
|
||||
"repository": repository.__class__.__name__,
|
||||
}
|
||||
if hasattr(repository, "ping"):
|
||||
repository.ping()
|
||||
except Exception as exc:
|
||||
app_context.console.print(f"health-check failed: {exc}")
|
||||
raise typer.Exit(code=1) from exc
|
||||
else:
|
||||
if app_context.output_format == "json":
|
||||
print_json(app_context.console, health)
|
||||
else:
|
||||
for key, value in health.items():
|
||||
app_context.console.print(f"{key}: {value}")
|
||||
finally:
|
||||
if hasattr(repository, "close"):
|
||||
repository.close()
|
||||
@@ -0,0 +1,262 @@
|
||||
"""Portfolio CLI commands."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from typing import Any
|
||||
|
||||
import typer
|
||||
|
||||
from app.cli.core.utils import execute_command, render_output
|
||||
from app.schemas.portfolio import (
|
||||
BuyRequest,
|
||||
CreatePortfolioRequest,
|
||||
InvestRequest,
|
||||
ResetPortfolioRequest,
|
||||
SellRequest,
|
||||
TransferRequest,
|
||||
UpdatePortfolioRequest,
|
||||
)
|
||||
from app.services.portfolio_service import PortfolioService
|
||||
|
||||
|
||||
portfolio_app = typer.Typer(help="Portfolio lifecycle and trade commands.")
|
||||
|
||||
|
||||
def _parse_strategy_params(value: str) -> dict[str, Any]:
|
||||
if not value.strip():
|
||||
return {}
|
||||
parsed = json.loads(value)
|
||||
if not isinstance(parsed, dict):
|
||||
raise ValueError("strategy params must be a JSON object")
|
||||
return parsed
|
||||
|
||||
|
||||
def _with_service(ctx: typer.Context, callback):
|
||||
app_context = ctx.obj
|
||||
repository = app_context.repository_factory(app_context.settings)
|
||||
service = PortfolioService(repository)
|
||||
try:
|
||||
result = execute_command(app_context.console, callback=lambda: callback(service, app_context))
|
||||
return result
|
||||
finally:
|
||||
if hasattr(repository, "close"):
|
||||
repository.close()
|
||||
|
||||
|
||||
def _confirm(force: bool, message: str) -> None:
|
||||
if force:
|
||||
return
|
||||
confirmed = typer.confirm(message)
|
||||
if not confirmed:
|
||||
raise typer.Exit(code=0)
|
||||
|
||||
|
||||
@portfolio_app.command("list")
|
||||
def list_portfolios(
|
||||
ctx: typer.Context,
|
||||
broker: str | None = typer.Option(None, help="Filter by broker."),
|
||||
status: str | None = typer.Option(None, help="Filter by status."),
|
||||
) -> None:
|
||||
detail = _with_service(ctx, lambda service, app_context: service.list_portfolios(broker=broker, status=status))
|
||||
render_output(ctx.obj.console, ctx.obj.output_format, detail, empty_message="No portfolios found.")
|
||||
|
||||
|
||||
@portfolio_app.command("status")
|
||||
def portfolio_status(
|
||||
ctx: typer.Context,
|
||||
portfolio_id: str = typer.Argument(..., help="Portfolio identifier."),
|
||||
recent: int = typer.Option(5, "--recent", min=1, help="Number of recent ledger entries."),
|
||||
) -> None:
|
||||
detail = _with_service(
|
||||
ctx,
|
||||
lambda service, app_context: service.get_portfolio_status(portfolio_id=portfolio_id, recent_trades=recent),
|
||||
)
|
||||
render_output(ctx.obj.console, ctx.obj.output_format, detail)
|
||||
|
||||
|
||||
@portfolio_app.command("create")
|
||||
def create_portfolio(
|
||||
ctx: typer.Context,
|
||||
portfolio_id: str = typer.Argument(..., help="Portfolio identifier."),
|
||||
symbol: str = typer.Option(..., help="Trading symbol."),
|
||||
broker: str = typer.Option("kis", help="Broker code."),
|
||||
status: str = typer.Option("enabled", help="Portfolio status."),
|
||||
cash_balance: float = typer.Option(0.0, min=0.0, help="Starting cash balance."),
|
||||
asset_quantity: float = typer.Option(0.0, min=0.0, help="Starting asset quantity."),
|
||||
asset_price: float = typer.Option(0.0, min=0.0, help="Starting asset price."),
|
||||
profit_buffer: float = typer.Option(0.0, min=0.0, help="Initial profit buffer."),
|
||||
strategy_name: str = typer.Option("target_value", help="Strategy name."),
|
||||
strategy_params: str = typer.Option("{}", help="Strategy params as JSON object."),
|
||||
force: bool = typer.Option(False, "--force", help="Skip confirmation prompt."),
|
||||
) -> None:
|
||||
_confirm(force, f"Create portfolio {portfolio_id} for symbol {symbol}?")
|
||||
detail = _with_service(
|
||||
ctx,
|
||||
lambda service, app_context: service.create_portfolio(
|
||||
CreatePortfolioRequest(
|
||||
portfolio_id=portfolio_id,
|
||||
symbol=symbol,
|
||||
broker=broker,
|
||||
status=status,
|
||||
cash_balance=cash_balance,
|
||||
asset_quantity=asset_quantity,
|
||||
asset_price=asset_price,
|
||||
profit_buffer=profit_buffer,
|
||||
strategy_name=strategy_name,
|
||||
strategy_params=_parse_strategy_params(strategy_params),
|
||||
)
|
||||
),
|
||||
)
|
||||
render_output(ctx.obj.console, ctx.obj.output_format, detail)
|
||||
|
||||
|
||||
@portfolio_app.command("update")
|
||||
def update_portfolio(
|
||||
ctx: typer.Context,
|
||||
portfolio_id: str = typer.Argument(..., help="Portfolio identifier."),
|
||||
broker: str | None = typer.Option(None, help="Broker code."),
|
||||
status: str | None = typer.Option(None, help="Portfolio status."),
|
||||
symbol: str | None = typer.Option(None, help="Trading symbol."),
|
||||
strategy_name: str | None = typer.Option(None, help="Strategy name."),
|
||||
strategy_params: str | None = typer.Option(None, help="Strategy params as JSON object."),
|
||||
asset_price: float | None = typer.Option(None, min=0.0, help="Current asset price."),
|
||||
force: bool = typer.Option(False, "--force", help="Skip confirmation prompt."),
|
||||
) -> None:
|
||||
_confirm(force, f"Update portfolio {portfolio_id}?")
|
||||
parsed_strategy_params = None if strategy_params is None else _parse_strategy_params(strategy_params)
|
||||
detail = _with_service(
|
||||
ctx,
|
||||
lambda service, app_context: service.update_portfolio(
|
||||
portfolio_id,
|
||||
UpdatePortfolioRequest(
|
||||
broker=broker,
|
||||
status=status,
|
||||
symbol=symbol,
|
||||
strategy_name=strategy_name,
|
||||
strategy_params=parsed_strategy_params,
|
||||
asset_price=asset_price,
|
||||
),
|
||||
),
|
||||
)
|
||||
render_output(ctx.obj.console, ctx.obj.output_format, detail)
|
||||
|
||||
|
||||
@portfolio_app.command("buy")
|
||||
def buy_portfolio(
|
||||
ctx: typer.Context,
|
||||
portfolio_id: str = typer.Argument(..., help="Portfolio identifier."),
|
||||
quantity: float = typer.Option(..., min=0.0000001, help="Quantity to buy."),
|
||||
price: float = typer.Option(..., min=0.0000001, help="Execution price."),
|
||||
fee: float = typer.Option(0.0, min=0.0, help="Fee amount."),
|
||||
tax: float = typer.Option(0.0, min=0.0, help="Tax amount."),
|
||||
note: str = typer.Option("", help="Ledger note."),
|
||||
force: bool = typer.Option(False, "--force", help="Skip confirmation prompt."),
|
||||
) -> None:
|
||||
total_cost = quantity * price + fee + tax
|
||||
_confirm(force, f"Buy {quantity:g} units of {portfolio_id} at {price:,.0f} (total {total_cost:,.0f})?")
|
||||
detail = _with_service(
|
||||
ctx,
|
||||
lambda service, app_context: service.buy(
|
||||
portfolio_id,
|
||||
BuyRequest(quantity=quantity, price=price, fee=fee, tax=tax, note=note),
|
||||
),
|
||||
)
|
||||
render_output(ctx.obj.console, ctx.obj.output_format, detail)
|
||||
|
||||
|
||||
@portfolio_app.command("sell")
|
||||
def sell_portfolio(
|
||||
ctx: typer.Context,
|
||||
portfolio_id: str = typer.Argument(..., help="Portfolio identifier."),
|
||||
quantity: float = typer.Option(..., min=0.0000001, help="Quantity to sell."),
|
||||
price: float = typer.Option(..., min=0.0000001, help="Execution price."),
|
||||
fee: float = typer.Option(0.0, min=0.0, help="Fee amount."),
|
||||
tax: float = typer.Option(0.0, min=0.0, help="Tax amount."),
|
||||
note: str = typer.Option("", help="Ledger note."),
|
||||
force: bool = typer.Option(False, "--force", help="Skip confirmation prompt."),
|
||||
) -> None:
|
||||
net_amount = quantity * price - fee - tax
|
||||
_confirm(force, f"Sell {quantity:g} units of {portfolio_id} at {price:,.0f} (net {net_amount:,.0f})?")
|
||||
detail = _with_service(
|
||||
ctx,
|
||||
lambda service, app_context: service.sell(
|
||||
portfolio_id,
|
||||
SellRequest(quantity=quantity, price=price, fee=fee, tax=tax, note=note),
|
||||
),
|
||||
)
|
||||
render_output(ctx.obj.console, ctx.obj.output_format, detail)
|
||||
|
||||
|
||||
@portfolio_app.command("invest")
|
||||
def invest_portfolio(
|
||||
ctx: typer.Context,
|
||||
portfolio_id: str = typer.Argument(..., help="Portfolio identifier."),
|
||||
movement_type: str = typer.Option(..., "--type", help="deposit, profit, or withdraw."),
|
||||
amount: float = typer.Option(..., min=0.0000001, help="Buffer adjustment amount."),
|
||||
note: str = typer.Option("", help="Ledger note."),
|
||||
force: bool = typer.Option(False, "--force", help="Skip confirmation prompt."),
|
||||
) -> None:
|
||||
_confirm(force, f"Apply {movement_type} of {amount:,.0f} to portfolio {portfolio_id}?")
|
||||
detail = _with_service(
|
||||
ctx,
|
||||
lambda service, app_context: service.invest(
|
||||
portfolio_id,
|
||||
InvestRequest(movement_type=movement_type, amount=amount, note=note),
|
||||
),
|
||||
)
|
||||
render_output(ctx.obj.console, ctx.obj.output_format, detail)
|
||||
|
||||
|
||||
@portfolio_app.command("transfer")
|
||||
def transfer_portfolio(
|
||||
ctx: typer.Context,
|
||||
portfolio_id: str = typer.Argument(..., help="Portfolio identifier."),
|
||||
direction: str = typer.Option(..., help="to-buffer or from-buffer."),
|
||||
amount: float = typer.Option(..., min=0.0000001, help="Transfer amount."),
|
||||
note: str = typer.Option("", help="Ledger note."),
|
||||
force: bool = typer.Option(False, "--force", help="Skip confirmation prompt."),
|
||||
) -> None:
|
||||
_confirm(force, f"Transfer {amount:,.0f} {direction} for portfolio {portfolio_id}?")
|
||||
detail = _with_service(
|
||||
ctx,
|
||||
lambda service, app_context: service.transfer(
|
||||
portfolio_id,
|
||||
TransferRequest(direction=direction, amount=amount, note=note),
|
||||
),
|
||||
)
|
||||
render_output(ctx.obj.console, ctx.obj.output_format, detail)
|
||||
|
||||
|
||||
@portfolio_app.command("reset")
|
||||
def reset_portfolio(
|
||||
ctx: typer.Context,
|
||||
portfolio_id: str = typer.Argument(..., help="Portfolio identifier."),
|
||||
infer_symbol: str | None = typer.Option(None, help="Fallback symbol when portfolio does not exist."),
|
||||
force: bool = typer.Option(False, "--force", help="Skip confirmation prompt."),
|
||||
) -> None:
|
||||
_confirm(force, f"Reset portfolio {portfolio_id}?")
|
||||
detail = _with_service(
|
||||
ctx,
|
||||
lambda service, app_context: service.reset(portfolio_id, ResetPortfolioRequest(infer_symbol=infer_symbol)),
|
||||
)
|
||||
render_output(ctx.obj.console, ctx.obj.output_format, detail)
|
||||
|
||||
|
||||
@portfolio_app.command("delete")
|
||||
def delete_portfolio(
|
||||
ctx: typer.Context,
|
||||
portfolio_id: str = typer.Argument(..., help="Portfolio identifier."),
|
||||
keep_ledger: bool = typer.Option(False, help="Preserve ledger history."),
|
||||
force: bool = typer.Option(False, "--force", help="Skip confirmation prompt."),
|
||||
) -> None:
|
||||
suffix = " and keep ledger" if keep_ledger else ""
|
||||
_confirm(force, f"Delete portfolio {portfolio_id}{suffix}?")
|
||||
_with_service(
|
||||
ctx,
|
||||
lambda service, app_context: service.delete(portfolio_id, keep_ledger=keep_ledger),
|
||||
)
|
||||
if ctx.obj.output_format == "json":
|
||||
render_output(ctx.obj.console, ctx.obj.output_format, {"deleted": portfolio_id, "keep_ledger": keep_ledger})
|
||||
else:
|
||||
ctx.obj.console.print(f"Deleted portfolio: {portfolio_id}")
|
||||
@@ -0,0 +1 @@
|
||||
"""CLI helpers."""
|
||||
@@ -0,0 +1,161 @@
|
||||
"""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.portfolio import PortfolioDetailResponse, PortfolioListItem
|
||||
|
||||
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:
|
||||
table = Table(title="Portfolios", expand=True)
|
||||
table.add_column("ID", style="cyan", no_wrap=True)
|
||||
table.add_column("Broker")
|
||||
table.add_column("Status", no_wrap=True)
|
||||
table.add_column("Symbol", no_wrap=True)
|
||||
table.add_column("Cash", justify="right")
|
||||
table.add_column("Qty", justify="right")
|
||||
table.add_column("Price", justify="right")
|
||||
table.add_column("Buffer", justify="right")
|
||||
table.add_column("Total", justify="right")
|
||||
for item in portfolios:
|
||||
table.add_row(
|
||||
item.portfolio_id,
|
||||
item.broker,
|
||||
item.status,
|
||||
item.symbol,
|
||||
format_currency(item.cash_balance),
|
||||
f"{item.asset_quantity:g}",
|
||||
format_currency(item.price),
|
||||
format_currency(item.profit_buffer),
|
||||
format_currency(item.total_value),
|
||||
)
|
||||
console.print(table)
|
||||
|
||||
|
||||
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"strategy: {portfolio.strategy_name}",
|
||||
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.strategy_name}",
|
||||
f"config: {detail.strategy_set.strategy_config}",
|
||||
]
|
||||
),
|
||||
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_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
|
||||
print_portfolio_table(console, payload)
|
||||
return
|
||||
|
||||
if isinstance(payload, PortfolioDetailResponse):
|
||||
print_portfolio_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
|
||||
@@ -0,0 +1,56 @@
|
||||
"""Typer entrypoint for PortfolioHub."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from typing import Callable
|
||||
|
||||
import typer
|
||||
from rich.console import Console
|
||||
|
||||
from app.cli.commands.health import register_health_commands
|
||||
from app.cli.commands.portfolio import portfolio_app
|
||||
from app.core.config import Settings, get_settings
|
||||
from app.repositories.factory import get_portfolio_repository_from_settings
|
||||
from app.repositories.portfolio_repository import PortfolioRepository
|
||||
|
||||
|
||||
@dataclass
|
||||
class AppContext:
|
||||
settings: Settings
|
||||
output_format: str
|
||||
console: Console
|
||||
repository_factory: Callable[[Settings], PortfolioRepository]
|
||||
|
||||
|
||||
def create_app(
|
||||
*,
|
||||
settings_factory: Callable[[], Settings] = get_settings,
|
||||
repository_factory: Callable[[Settings], PortfolioRepository] = get_portfolio_repository_from_settings,
|
||||
) -> typer.Typer:
|
||||
app = typer.Typer(help="PortfolioHub command line interface.", no_args_is_help=True)
|
||||
|
||||
@app.callback()
|
||||
def main_callback(
|
||||
ctx: typer.Context,
|
||||
output_format: str = typer.Option("table", "--format", help="Output format: table or json."),
|
||||
) -> None:
|
||||
settings = settings_factory()
|
||||
console = Console(no_color=True, width=160)
|
||||
ctx.obj = AppContext(
|
||||
settings=settings,
|
||||
output_format=output_format,
|
||||
console=console,
|
||||
repository_factory=repository_factory,
|
||||
)
|
||||
|
||||
app.add_typer(portfolio_app, name="portfolio")
|
||||
register_health_commands(app)
|
||||
return app
|
||||
|
||||
|
||||
app = create_app()
|
||||
|
||||
|
||||
def main() -> None:
|
||||
app()
|
||||
Reference in New Issue
Block a user