124 lines
4.3 KiB
Python
124 lines
4.3 KiB
Python
"""User CLI commands."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import typer
|
|
|
|
from app.cli.core.utils import execute_command, render_output
|
|
from app.cli.commands.portfolio import _confirm
|
|
from app.schemas.user import CreateUserRequest, UpdateUserRequest, UserResponse
|
|
from app.services.user_service import UserService
|
|
|
|
|
|
user_app = typer.Typer(help="User management commands.")
|
|
|
|
|
|
def _with_service(ctx: typer.Context, callback):
|
|
app_context = ctx.obj
|
|
repository = app_context.repository_factory(app_context.settings)
|
|
service = UserService(repository)
|
|
try:
|
|
return execute_command(app_context.console, callback=lambda: callback(service, app_context))
|
|
finally:
|
|
if callable(getattr(type(repository), "close", None)):
|
|
repository.close()
|
|
|
|
|
|
def _to_user_response(user) -> UserResponse:
|
|
return UserResponse(
|
|
id=user.id,
|
|
username=user.username,
|
|
email=user.email,
|
|
permission=user.permission,
|
|
is_active=user.is_active,
|
|
created_at=user.created_at,
|
|
updated_at=user.updated_at,
|
|
last_login_at=user.last_login_at,
|
|
)
|
|
|
|
|
|
@user_app.command("list")
|
|
def list_users(ctx: typer.Context) -> None:
|
|
users = _with_service(ctx, lambda service, app_context: [_to_user_response(user) for user in service.list()])
|
|
render_output(ctx.obj.console, ctx.obj.output_format, users, empty_message="No users found.")
|
|
|
|
|
|
@user_app.command("get")
|
|
def get_user(
|
|
ctx: typer.Context,
|
|
user_id: str = typer.Argument(..., help="User identifier."),
|
|
) -> None:
|
|
user = _with_service(ctx, lambda service, app_context: _to_user_response(service.get(user_id)))
|
|
render_output(ctx.obj.console, ctx.obj.output_format, user)
|
|
|
|
|
|
@user_app.command("create")
|
|
def create_user(
|
|
ctx: typer.Context,
|
|
username: str = typer.Option(..., help="Username."),
|
|
password: str = typer.Option(..., help="Initial password."),
|
|
email: str | None = typer.Option(None, help="Email address."),
|
|
permission: str = typer.Option("read", help="Permission: read, trade-write, admin."),
|
|
is_active: bool = typer.Option(True, help="Whether the user is active."),
|
|
force: bool = typer.Option(False, "--force", help="Skip confirmation prompt."),
|
|
) -> None:
|
|
_confirm(force, f"Create user {username}?")
|
|
user = _with_service(
|
|
ctx,
|
|
lambda service, app_context: _to_user_response(
|
|
service.create(
|
|
CreateUserRequest(
|
|
username=username,
|
|
password=password,
|
|
email=email,
|
|
permission=permission,
|
|
is_active=is_active,
|
|
)
|
|
)
|
|
),
|
|
)
|
|
render_output(ctx.obj.console, ctx.obj.output_format, user)
|
|
|
|
|
|
@user_app.command("update")
|
|
def update_user(
|
|
ctx: typer.Context,
|
|
user_id: str = typer.Argument(..., help="User identifier."),
|
|
username: str | None = typer.Option(None, help="New username."),
|
|
password: str | None = typer.Option(None, help="New password."),
|
|
email: str | None = typer.Option(None, help="New email address."),
|
|
permission: str | None = typer.Option(None, help="New permission."),
|
|
is_active: bool | None = typer.Option(None, help="Whether the user is active."),
|
|
force: bool = typer.Option(False, "--force", help="Skip confirmation prompt."),
|
|
) -> None:
|
|
_confirm(force, f"Update user {user_id}?")
|
|
user = _with_service(
|
|
ctx,
|
|
lambda service, app_context: _to_user_response(
|
|
service.update(
|
|
user_id,
|
|
UpdateUserRequest(
|
|
username=username,
|
|
password=password,
|
|
email=email,
|
|
permission=permission,
|
|
is_active=is_active,
|
|
),
|
|
)
|
|
),
|
|
)
|
|
render_output(ctx.obj.console, ctx.obj.output_format, user)
|
|
|
|
|
|
@user_app.command("delete")
|
|
def delete_user(
|
|
ctx: typer.Context,
|
|
user_id: str = typer.Argument(..., help="User identifier."),
|
|
force: bool = typer.Option(False, "--force", help="Skip confirmation prompt."),
|
|
) -> None:
|
|
_confirm(force, f"Delete user {user_id}?")
|
|
_with_service(ctx, lambda service, app_context: service.delete(user_id))
|
|
if ctx.obj.output_format == "json":
|
|
render_output(ctx.obj.console, ctx.obj.output_format, {"deleted": user_id})
|
|
else:
|
|
ctx.obj.console.print(f"Deleted user: {user_id}") |