Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 0 additions & 4 deletions docs/cli.md
Original file line number Diff line number Diff line change
@@ -1,7 +1,5 @@
# `pyaxm-cli`

Query Apple Business Manager using Python.

**Usage**:

```console
Expand All @@ -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.
Expand Down
175 changes: 67 additions & 108 deletions pyaxm/cli.py
Original file line number Diff line number Diff line change
@@ -1,186 +1,133 @@
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 ────────────────────────────────────────────────────


@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,
Expand All @@ -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__":
Expand Down
1 change: 0 additions & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,6 @@ dependencies = [
"pyjwt",
"pandas",
"typer",
"rich",
]
readme = "README.md"
license = "GPL-3.0-or-later"
Expand Down
1 change: 0 additions & 1 deletion requirements.txt
Original file line number Diff line number Diff line change
Expand Up @@ -3,4 +3,3 @@ requests
pyjwt
pandas
typer
rich
Loading