From d40e9642989e11ebb0c9138f7f4ba85764be0f96 Mon Sep 17 00:00:00 2001 From: Kain Date: Sat, 11 Jul 2026 20:57:40 -0700 Subject: [PATCH] feat(cli): replace raw CSV output with rich tables and add format/output flags - Add `--format` flag to switch between table (default) and csv output - Add `--output` / `-o` flag to save results to a file with auto-naming for directories - Auto-detect pipe/redirect and switch to CSV so scripts still work - Introduce `build_records()` and `render_output()` helpers to eliminate boilerplate - Add `rich` to project dependencies - Regenerate CLI docs --- docs/cli.md | 4 + pyaxm/cli.py | 225 +++++++++++++++++++++++++++++++---------------- pyproject.toml | 1 + requirements.txt | 3 +- 4 files changed, 157 insertions(+), 76 deletions(-) diff --git a/docs/cli.md b/docs/cli.md index 8fdea17..e898620 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -1,5 +1,7 @@ # `pyaxm-cli` +Query Apple Business Manager using Python. + **Usage**: ```console @@ -8,6 +10,8 @@ $ pyaxm-cli [OPTIONS] COMMAND [ARGS]... **Options**: +* `--format TEXT`: Output format: table or csv [default: table] +* `-o, --output TEXT`: Save output to file or directory (auto-names inside directories) * `--install-completion`: Install completion for the current shell. * `--show-completion`: Show completion for the current shell, to copy it or customize the installation. * `--help`: Show this message and exit. diff --git a/pyaxm/cli.py b/pyaxm/cli.py index 8716b4c..63403f0 100644 --- a/pyaxm/cli.py +++ b/pyaxm/cli.py @@ -1,106 +1,186 @@ import sys +import os +from datetime import date import pandas as pd import typer from typing_extensions import Annotated from typing import List, Optional from pyaxm.client import Client from pyaxm.utils import download_activity_csv +from rich.console import Console +from rich.table import Table app = typer.Typer() +console = Console() + + +def build_records(items): + """Build a list of flat dicts from API response items.""" + records = [] + for item in items: + record = {"id": item.id} + if hasattr(item, "attributes") and item.attributes: + record.update(item.attributes.model_dump()) + records.append(record) + return records + + +def render_output(records, fmt, output_path, cmd_name): + """Render records as a rich table or CSV. + + Auto-switches to CSV when stdout is piped/redirected and no + explicit format was set. + """ + if not records: + console.print("[yellow]No results found.[/yellow]") + return + + columns = list(records[0].keys()) + df = pd.DataFrame(records, columns=columns) + + # --output / -o flag: save to file (auto-name inside directories) + if output_path: + if os.path.isdir(output_path): + filename = f"pyaxm-{cmd_name}-{date.today().isoformat()}.csv" + output_path = os.path.join(output_path, filename) + df.to_csv(output_path, index=False) + console.print(f"[green]Saved to: {output_path}[/green]") + return + + # When piped / redirected, default to CSV for scripts + if fmt == "table" and not sys.stdout.isatty(): + fmt = "csv" + + if fmt == "csv": + df.to_csv(sys.stdout, index=False) + else: + table = Table(show_header=True, header_style="bold cyan") + for col in columns: + table.add_column(str(col)) + for _, row in df.iterrows(): + table.add_row(*[str(v) if pd.notna(v) else "" for v in row]) + console.print(table) + + +# ── global options ────────────────────────────────────────────────── + + +@app.callback() +def main( + ctx: typer.Context, + format: Annotated[str, typer.Option("--format", help="Output format: table or csv")] = "table", + output: Annotated[ + Optional[str], + typer.Option("--output", "-o", help="Save output to file or directory (auto-names inside directories)"), + ] = None, +): + """Query Apple Business Manager using Python.""" + ctx.obj = {"format": format, "output": output} + + +# ── device commands ───────────────────────────────────────────────── + @app.command() -def devices(): +def devices(ctx: typer.Context): """List all devices in the organization.""" client = Client() - devices = client.list_devices() - devices_data = [] - for device in devices: - device_info = {'id': device.id} - device_info.update(device.attributes.model_dump()) - devices_data.append(device_info) - df = pd.DataFrame(devices_data) - df.to_csv(sys.stdout, index=False) + records = build_records(client.list_devices()) + render_output(records, ctx.obj["format"], ctx.obj["output"], "devices") + @app.command() -def device(device_id: Annotated[str, typer.Argument()]): +def device(ctx: typer.Context, device_id: Annotated[str, typer.Argument()]): """Get a device by ID.""" client = Client() - device = client.get_device(device_id) - device_info = {'id': device.id} - device_info.update(device.attributes.model_dump()) - df = pd.DataFrame([device_info]) - df.to_csv(sys.stdout, index=False) + item = client.get_device(device_id) + record = {"id": item.id} + if item.attributes: + record.update(item.attributes.model_dump()) + render_output([record], ctx.obj["format"], ctx.obj["output"], "device") + @app.command() -def apple_care_coverage(device_id: Annotated[str, typer.Argument()]): +def apple_care_coverage(ctx: typer.Context, device_id: Annotated[str, typer.Argument()]): """Get AppleCare coverage for a device.""" client = Client() - coverage = client.get_apple_care_coverage(device_id) - coverage_data = [item.attributes.model_dump() for item in coverage] - df = pd.DataFrame(coverage_data) - df.to_csv(sys.stdout, index=False) + records = build_records(client.get_apple_care_coverage(device_id)) + render_output(records, ctx.obj["format"], ctx.obj["output"], "apple-care-coverage") + + +# ── MDM server commands ───────────────────────────────────────────── + @app.command() -def mdm_servers(): +def mdm_servers(ctx: typer.Context): """List all MDM servers.""" client = Client() - servers = client.list_mdm_servers() - servers_data = [] - for server in servers: - server_info = {'id': server.id} - server_info.update(server.attributes.model_dump()) - servers_data.append(server_info) - df = pd.DataFrame(servers_data) - df.to_csv(sys.stdout, index=False) + records = build_records(client.list_mdm_servers()) + render_output(records, ctx.obj["format"], ctx.obj["output"], "mdm-servers") + @app.command() -def mdm_server(server_id: Annotated[str, typer.Argument()]): +def mdm_server(ctx: typer.Context, server_id: Annotated[str, typer.Argument()]): """List devices in a specific MDM server.""" client = Client() devices = client.list_devices_in_mdm_server(server_id) - devices_data = [{'id': device.id} for device in devices] - df = pd.DataFrame(devices_data) - df.to_csv(sys.stdout, index=False) + records = [{"id": d.id} for d in devices] + render_output(records, ctx.obj["format"], ctx.obj["output"], "mdm-server") + @app.command() -def mdm_server_assigned(device_id: Annotated[str, typer.Argument()]): +def mdm_server_assigned(ctx: typer.Context, device_id: Annotated[str, typer.Argument()]): """Get the server assignment for a device.""" client = Client() - server_assignment = client.get_device_server_assignment(device_id) - assignment_info = {'id': server_assignment.id} - df = pd.DataFrame([assignment_info]) - df.to_csv(sys.stdout, index=False) + assignment = client.get_device_server_assignment(device_id) + records = [{"id": assignment.id}] + render_output(records, ctx.obj["format"], ctx.obj["output"], "mdm-server-assigned") + @app.command() -def assign_device(device_ids: Annotated[List[str], typer.Argument()], server_id: Annotated[str, typer.Argument()]): +def assign_device( + ctx: typer.Context, + device_ids: Annotated[List[str], typer.Argument()], + server_id: Annotated[str, typer.Argument()], +): """Assign one or more devices to an MDM server.""" client = Client() - activity = client.assign_unassign_device_to_mdm_server(device_ids, server_id, 'ASSIGN_DEVICES') - activity_data = {'id': activity.id} - activity_data.update(activity.attributes.model_dump()) - df = pd.DataFrame([activity_data]) - df.to_csv(sys.stdout, index=False) + activity = client.assign_unassign_device_to_mdm_server(device_ids, server_id, "ASSIGN_DEVICES") + record = {"id": activity.id} + if activity.attributes: + record.update(activity.attributes.model_dump()) + render_output([record], ctx.obj["format"], ctx.obj["output"], "assign-device") file_path = download_activity_csv(activity) if file_path: - typer.echo(f"Report downloaded successfully to: {file_path}") + console.print(f"[green]Report downloaded: {file_path}[/green]") + @app.command() -def unassign_device(device_ids: Annotated[List[str], typer.Argument()], server_id: Annotated[str, typer.Argument()]): +def unassign_device( + ctx: typer.Context, + device_ids: Annotated[List[str], typer.Argument()], + server_id: Annotated[str, typer.Argument()], +): """Unassign one or more devices from an MDM server.""" client = Client() - activity = client.assign_unassign_device_to_mdm_server(device_ids, server_id, 'UNASSIGN_DEVICES') - activity_data = {'id': activity.id} - activity_data.update(activity.attributes.model_dump()) - df = pd.DataFrame([activity_data]) - df.to_csv(sys.stdout, index=False) - + activity = client.assign_unassign_device_to_mdm_server(device_ids, server_id, "UNASSIGN_DEVICES") + record = {"id": activity.id} + if activity.attributes: + record.update(activity.attributes.model_dump()) + render_output([record], ctx.obj["format"], ctx.obj["output"], "unassign-device") + file_path = download_activity_csv(activity) if file_path: - typer.echo(f"Report downloaded successfully to: {file_path}") + console.print(f"[green]Report downloaded: {file_path}[/green]") + + +# ── audit events ──────────────────────────────────────────────────── + @app.command() def audit_events( + ctx: typer.Context, start_timestamp: Annotated[str, typer.Argument()], end_timestamp: Annotated[str, typer.Argument()], actor_id: Annotated[Optional[str], typer.Option("--actor-id", "-a")] = None, @@ -122,36 +202,31 @@ def audit_events( fields=fields, cursor=cursor, ) - events_data = [] - for event in events: - event_info = {'id': event.id} - event_info.update(event.attributes.model_dump()) - events_data.append(event_info) - df = pd.DataFrame(events_data) - df.to_csv(sys.stdout, index=False) + records = build_records(events) + render_output(records, ctx.obj["format"], ctx.obj["output"], "audit-events") + + +# ── user commands ─────────────────────────────────────────────────── + @app.command() -def users(): +def users(ctx: typer.Context): """List all users in the organization.""" client = Client() - users = client.list_users() - users_data = [] - for user in users: - user_info = {'id': user.id} - user_info.update(user.attributes.model_dump()) - users_data.append(user_info) - df = pd.DataFrame(users_data) - df.to_csv(sys.stdout, index=False) + records = build_records(client.list_users()) + render_output(records, ctx.obj["format"], ctx.obj["output"], "users") + @app.command() -def user(user_id: Annotated[str, typer.Argument()]): +def user(ctx: typer.Context, user_id: Annotated[str, typer.Argument()]): """Get a user by ID.""" client = Client() - user = client.get_user(user_id) - user_info = {'id': user.id} - user_info.update(user.attributes.model_dump()) - df = pd.DataFrame([user_info]) - df.to_csv(sys.stdout, index=False) + item = client.get_user(user_id) + record = {"id": item.id} + if item.attributes: + record.update(item.attributes.model_dump()) + render_output([record], ctx.obj["format"], ctx.obj["output"], "user") + if __name__ == "__main__": app() diff --git a/pyproject.toml b/pyproject.toml index 5baf51d..398750f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -13,6 +13,7 @@ dependencies = [ "pyjwt", "pandas", "typer", + "rich", ] readme = "README.md" license = "GPL-3.0-or-later" diff --git a/requirements.txt b/requirements.txt index 59ca3a5..d7a8d55 100644 --- a/requirements.txt +++ b/requirements.txt @@ -2,4 +2,5 @@ pydantic requests pyjwt pandas -typer \ No newline at end of file +typer +rich \ No newline at end of file