genichin
222c66bb7f
- Add CredentialService for managing broker credentials - Implement credential CLI commands (create, list, get, update, delete) - Add postgres repository support for credential operations - Add comprehensive test coverage for credential CLI - Update docs/batches/index.md and docs/srs/index.md with BAT-005 - Update docs/project-structure.md to reflect credential service
128 lines
4.8 KiB
Python
128 lines
4.8 KiB
Python
"""Broker credential CLI commands."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
import typer
|
|
|
|
from app.cli.commands.portfolio import _confirm
|
|
from app.cli.core.utils import execute_command, render_output
|
|
from app.core.encryption import encrypt_payload
|
|
|
|
credential_app = typer.Typer(help="Broker credential management commands.")
|
|
|
|
|
|
def _with_service(ctx: typer.Context, callback):
|
|
app_context = ctx.obj
|
|
repository = app_context.repository_factory(app_context.settings)
|
|
try:
|
|
return execute_command(app_context.console, callback=lambda: callback(repository, app_context))
|
|
except SystemExit:
|
|
raise
|
|
except RuntimeError as exc:
|
|
app_context.console.print(str(exc))
|
|
raise typer.Exit(code=1) from exc
|
|
finally:
|
|
if callable(getattr(type(repository), "close", None)):
|
|
repository.close()
|
|
|
|
|
|
@credential_app.command("set")
|
|
def set_credential(
|
|
ctx: typer.Context,
|
|
user: str = typer.Option(..., "--user", help="Owner username (required)."),
|
|
broker: str = typer.Option("kis", "--broker", help="Broker code (default: kis)."),
|
|
account_ref: str = typer.Option(..., "--account-ref", help="Account reference (e.g. account number)."),
|
|
app_key: str = typer.Option(..., "--app-key", help="Broker app_key."),
|
|
app_secret: str = typer.Option(..., "--app-secret", help="Broker app_secret."),
|
|
force: bool = typer.Option(False, "--force", help="Skip confirmation prompt."),
|
|
) -> None:
|
|
"""Register or update a broker credential (upsert)."""
|
|
_confirm(force, f"Set {broker} credential for user '{user}'?")
|
|
|
|
def _callback(repository, app_context):
|
|
owner = repository.get_user_by_username(user)
|
|
if owner is None:
|
|
raise ValueError(f"사용자를 찾을 수 없습니다: {user}")
|
|
enc_key = app_context.settings.require_credential_encryption_key()
|
|
payload = json.dumps({"app_key": app_key, "app_secret": app_secret})
|
|
encrypted = encrypt_payload(enc_key, payload)
|
|
return repository.upsert_broker_credential(
|
|
owner.id,
|
|
broker,
|
|
account_ref=account_ref,
|
|
encrypted_payload=encrypted,
|
|
)
|
|
|
|
result = _with_service(ctx, _callback)
|
|
if result is not None:
|
|
render_output(
|
|
ctx.obj.console,
|
|
ctx.obj.output_format,
|
|
{
|
|
"user_id": result.user_id,
|
|
"broker": result.broker,
|
|
"account_ref": result.account_ref,
|
|
"credential_version": result.credential_version,
|
|
"last_rotated_at": str(result.last_rotated_at),
|
|
},
|
|
)
|
|
|
|
|
|
@credential_app.command("list")
|
|
def list_credentials(
|
|
ctx: typer.Context,
|
|
user: str = typer.Option(..., "--user", help="Owner username (required)."),
|
|
) -> None:
|
|
"""List broker credentials for a user (payload not shown)."""
|
|
|
|
def _callback(repository, app_context):
|
|
owner = repository.get_user_by_username(user)
|
|
if owner is None:
|
|
raise ValueError(f"사용자를 찾을 수 없습니다: {user}")
|
|
records = repository.list_broker_credentials(owner.id)
|
|
return [
|
|
{
|
|
"broker": r.broker,
|
|
"account_ref": r.account_ref,
|
|
"credential_version": r.credential_version,
|
|
"last_rotated_at": str(r.last_rotated_at),
|
|
}
|
|
for r in records
|
|
]
|
|
|
|
result = _with_service(ctx, _callback)
|
|
if result is not None:
|
|
if ctx.obj.output_format == "json":
|
|
render_output(ctx.obj.console, ctx.obj.output_format, result)
|
|
else:
|
|
for item in result if isinstance(result, list) else [result]:
|
|
ctx.obj.console.print(item)
|
|
if not result:
|
|
ctx.obj.console.print("No broker credentials found.")
|
|
|
|
|
|
@credential_app.command("remove")
|
|
def remove_credential(
|
|
ctx: typer.Context,
|
|
user: str = typer.Option(..., "--user", help="Owner username (required)."),
|
|
broker: str = typer.Option("kis", "--broker", help="Broker code (default: kis)."),
|
|
force: bool = typer.Option(False, "--force", help="Skip confirmation prompt."),
|
|
) -> None:
|
|
"""Remove a broker credential."""
|
|
_confirm(force, f"Remove {broker} credential for user '{user}'? This cannot be undone.")
|
|
|
|
def _callback(repository, app_context):
|
|
owner = repository.get_user_by_username(user)
|
|
if owner is None:
|
|
raise ValueError(f"사용자를 찾을 수 없습니다: {user}")
|
|
deleted = repository.delete_broker_credential(owner.id, broker)
|
|
if not deleted:
|
|
raise ValueError(f"인증 정보를 찾을 수 없습니다: user={user} broker={broker}")
|
|
return {"deleted": True, "user": user, "broker": broker}
|
|
|
|
result = _with_service(ctx, _callback)
|
|
if result is not None:
|
|
render_output(ctx.obj.console, ctx.obj.output_format, result)
|