diff --git a/docs/cli.md b/docs/cli.md index e898620..8fdea17 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -1,7 +1,5 @@ # `pyaxm-cli` -Query Apple Business Manager using Python. - **Usage**: ```console @@ -10,8 +8,6 @@ $ 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 63403f0..0e3c3e3 100644 --- a/pyaxm/cli.py +++ b/pyaxm/cli.py @@ -1,178 +1,126 @@ 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(ctx: typer.Context): +def devices(): """List all devices in the organization.""" client = Client() - records = build_records(client.list_devices()) - render_output(records, ctx.obj["format"], ctx.obj["output"], "devices") + devices = client.list_devices() + records = [] + for d in devices: + info = {"id": d.id} + info.update(d.attributes.model_dump()) + records.append(info) + df = pd.DataFrame(records) + df.to_csv(sys.stdout, index=False) @app.command() -def device(ctx: typer.Context, device_id: Annotated[str, typer.Argument()]): +def device(device_id: Annotated[str, typer.Argument()]): """Get a device by ID.""" client = Client() - 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") + d = client.get_device(device_id) + info = {"id": d.id} + if d.attributes: + info.update(d.attributes.model_dump()) + df = pd.DataFrame([info]) + df.to_csv(sys.stdout, index=False) @app.command() -def apple_care_coverage(ctx: typer.Context, device_id: Annotated[str, typer.Argument()]): +def apple_care_coverage(device_id: Annotated[str, typer.Argument()]): """Get AppleCare coverage for a device.""" client = Client() - records = build_records(client.get_apple_care_coverage(device_id)) - render_output(records, ctx.obj["format"], ctx.obj["output"], "apple-care-coverage") + coverage = client.get_apple_care_coverage(device_id) + records = [{"id": item.id, **item.attributes.model_dump()} for item in coverage] + df = pd.DataFrame(records) + df.to_csv(sys.stdout, index=False) # ── MDM server commands ───────────────────────────────────────────── @app.command() -def mdm_servers(ctx: typer.Context): +def mdm_servers(): """List all MDM servers.""" client = Client() - records = build_records(client.list_mdm_servers()) - render_output(records, ctx.obj["format"], ctx.obj["output"], "mdm-servers") + servers = client.list_mdm_servers() + records = [] + for s in servers: + info = {"id": s.id} + info.update(s.attributes.model_dump()) + records.append(info) + df = pd.DataFrame(records) + df.to_csv(sys.stdout, index=False) @app.command() -def mdm_server(ctx: typer.Context, server_id: Annotated[str, typer.Argument()]): +def mdm_server(server_id: Annotated[str, typer.Argument()]): """List devices in a specific MDM server.""" client = Client() devices = client.list_devices_in_mdm_server(server_id) records = [{"id": d.id} for d in devices] - render_output(records, ctx.obj["format"], ctx.obj["output"], "mdm-server") + df = pd.DataFrame(records) + df.to_csv(sys.stdout, index=False) @app.command() -def mdm_server_assigned(ctx: typer.Context, device_id: Annotated[str, typer.Argument()]): +def mdm_server_assigned(device_id: Annotated[str, typer.Argument()]): """Get the server assignment for a device.""" client = Client() assignment = client.get_device_server_assignment(device_id) records = [{"id": assignment.id}] - render_output(records, ctx.obj["format"], ctx.obj["output"], "mdm-server-assigned") + df = pd.DataFrame(records) + df.to_csv(sys.stdout, index=False) @app.command() 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") - record = {"id": activity.id} + info = {"id": activity.id} if activity.attributes: - record.update(activity.attributes.model_dump()) - render_output([record], ctx.obj["format"], ctx.obj["output"], "assign-device") + info.update(activity.attributes.model_dump()) + df = pd.DataFrame([info]) + df.to_csv(sys.stdout, index=False) file_path = download_activity_csv(activity) if file_path: - console.print(f"[green]Report downloaded: {file_path}[/green]") + typer.echo(f"Report downloaded successfully to: {file_path}") @app.command() 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") - record = {"id": activity.id} + info = {"id": activity.id} if activity.attributes: - record.update(activity.attributes.model_dump()) - render_output([record], ctx.obj["format"], ctx.obj["output"], "unassign-device") + info.update(activity.attributes.model_dump()) + df = pd.DataFrame([info]) + df.to_csv(sys.stdout, index=False) file_path = download_activity_csv(activity) if file_path: - console.print(f"[green]Report downloaded: {file_path}[/green]") + typer.echo(f"Report downloaded successfully to: {file_path}") # ── audit events ──────────────────────────────────────────────────── @@ -180,7 +128,6 @@ def unassign_device( @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, @@ -202,30 +149,42 @@ def audit_events( fields=fields, cursor=cursor, ) - records = build_records(events) - render_output(records, ctx.obj["format"], ctx.obj["output"], "audit-events") + records = [] + for event in events: + info = {"id": event.id} + info.update(event.attributes.model_dump()) + records.append(info) + df = pd.DataFrame(records) + df.to_csv(sys.stdout, index=False) # ── user commands ─────────────────────────────────────────────────── @app.command() -def users(ctx: typer.Context): +def users(): """List all users in the organization.""" client = Client() - records = build_records(client.list_users()) - render_output(records, ctx.obj["format"], ctx.obj["output"], "users") + users = client.list_users() + records = [] + for u in users: + info = {"id": u.id} + info.update(u.attributes.model_dump()) + records.append(info) + df = pd.DataFrame(records) + df.to_csv(sys.stdout, index=False) @app.command() -def user(ctx: typer.Context, user_id: Annotated[str, typer.Argument()]): +def user(user_id: Annotated[str, typer.Argument()]): """Get a user by ID.""" client = Client() item = client.get_user(user_id) - record = {"id": item.id} + info = {"id": item.id} if item.attributes: - record.update(item.attributes.model_dump()) - render_output([record], ctx.obj["format"], ctx.obj["output"], "user") + info.update(item.attributes.model_dump()) + df = pd.DataFrame([info]) + df.to_csv(sys.stdout, index=False) if __name__ == "__main__": diff --git a/pyproject.toml b/pyproject.toml index 398750f..5baf51d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -13,7 +13,6 @@ dependencies = [ "pyjwt", "pandas", "typer", - "rich", ] readme = "README.md" license = "GPL-3.0-or-later" diff --git a/requirements.txt b/requirements.txt index d7a8d55..20a8eb7 100644 --- a/requirements.txt +++ b/requirements.txt @@ -3,4 +3,3 @@ requests pyjwt pandas typer -rich \ No newline at end of file