156 lines
4.8 KiB
Python
156 lines
4.8 KiB
Python
"""Main CLI module for api-snapshot-cli."""
|
|
|
|
import os
|
|
from typing import Optional
|
|
|
|
import click
|
|
from rich.console import Console
|
|
from rich.table import Table
|
|
|
|
from api_snapshot.snapshot.manager import SnapshotManager
|
|
|
|
console = Console()
|
|
|
|
|
|
def get_snapshot_dir() -> str:
|
|
"""Get the snapshot directory from environment or default."""
|
|
return os.environ.get("API_SNAPSHOT_DIR", "./snapshots")
|
|
|
|
|
|
def get_default_host() -> str:
|
|
"""Get the default host from environment."""
|
|
return os.environ.get("API_SNAPSHOT_HOST", "127.0.0.1")
|
|
|
|
|
|
def get_default_port() -> int:
|
|
"""Get the default port from environment."""
|
|
try:
|
|
return int(os.environ.get("API_SNAPSHOT_PORT", "8080"))
|
|
except ValueError:
|
|
return 8080
|
|
|
|
|
|
@click.group()
|
|
@click.version_option(version="0.1.0", prog_name="api-snapshot")
|
|
@click.option("--verbose", "-v", is_flag=True, help="Enable verbose output")
|
|
@click.option("--snapshot-dir", default=None, help="Override snapshot directory")
|
|
@click.pass_context
|
|
def main(ctx: click.Context, verbose: bool, snapshot_dir: Optional[str]) -> None:
|
|
"""API Snapshot CLI - Record HTTP API traffic and generate local mock servers.
|
|
|
|
This tool allows you to:
|
|
- Record HTTP requests and responses from any API
|
|
- Save them as JSON snapshots
|
|
- Spin up a local mock server to replay recorded responses
|
|
"""
|
|
ctx.ensure_object(dict)
|
|
ctx.obj["verbose"] = verbose
|
|
ctx.obj["snapshot_dir"] = snapshot_dir or get_snapshot_dir()
|
|
|
|
if verbose:
|
|
console.print("[cyan]Verbose mode enabled[/cyan]")
|
|
console.print(f"[cyan]Snapshot directory: {ctx.obj['snapshot_dir']}[/cyan]")
|
|
|
|
|
|
@main.command(name="list")
|
|
@click.pass_context
|
|
def list_snapshots(ctx: click.Context) -> None:
|
|
"""List all available snapshots."""
|
|
snapshot_dir = ctx.obj["snapshot_dir"]
|
|
manager = SnapshotManager(snapshot_dir)
|
|
|
|
snapshots = manager.list_snapshots()
|
|
|
|
if not snapshots:
|
|
console.print("[yellow]No snapshots found.[/yellow]")
|
|
return
|
|
|
|
table = Table(title="Available Snapshots")
|
|
table.add_column("Name", style="cyan")
|
|
table.add_column("Created", style="green")
|
|
table.add_column("Endpoints", style="magenta")
|
|
table.add_column("Size", style="blue")
|
|
|
|
for snapshot in snapshots:
|
|
table.add_row(
|
|
snapshot["name"],
|
|
snapshot["created"],
|
|
str(snapshot["endpoint_count"]),
|
|
snapshot["size"]
|
|
)
|
|
|
|
console.print(table)
|
|
|
|
|
|
@main.command(name="info")
|
|
@click.argument("name")
|
|
@click.pass_context
|
|
def snapshot_info(ctx: click.Context, name: str) -> None:
|
|
"""Show detailed information about a snapshot."""
|
|
snapshot_dir = ctx.obj["snapshot_dir"]
|
|
manager = SnapshotManager(snapshot_dir)
|
|
|
|
try:
|
|
snapshot = manager.load_snapshot(name)
|
|
except FileNotFoundError:
|
|
console.print(f"[red]Error: Snapshot '{name}' not found[/red]")
|
|
raise click.Abort()
|
|
except Exception as e:
|
|
console.print(f"[red]Error loading snapshot: {e}[/red]")
|
|
raise click.Abort()
|
|
|
|
table = Table(title=f"Snapshot: {name}")
|
|
table.add_column("Property", style="yellow")
|
|
table.add_column("Value", style="cyan")
|
|
|
|
version = getattr(snapshot.metadata, 'version', '1.0')
|
|
timestamp = getattr(snapshot.metadata, 'timestamp', 'Unknown')
|
|
description = getattr(snapshot.metadata, 'description', 'No description')
|
|
|
|
table.add_row("Version", version)
|
|
table.add_row("Created", timestamp)
|
|
table.add_row("Description", description)
|
|
table.add_row("Endpoints", str(len(snapshot.requests)))
|
|
|
|
console.print(table)
|
|
|
|
if snapshot.requests:
|
|
endpoint_table = Table(title="Recorded Endpoints")
|
|
endpoint_table.add_column("Method", style="cyan")
|
|
endpoint_table.add_column("Path", style="green")
|
|
endpoint_table.add_column("Status", style="magenta")
|
|
|
|
for req in snapshot.requests:
|
|
endpoint_table.add_row(
|
|
req.request.method,
|
|
req.request.url,
|
|
str(req.response.status_code)
|
|
)
|
|
|
|
console.print(endpoint_table)
|
|
|
|
|
|
@main.command(name="delete")
|
|
@click.argument("name")
|
|
@click.option("--force", "-f", is_flag=True, help="Skip confirmation")
|
|
@click.pass_context
|
|
def delete_snapshot(ctx: click.Context, name: str, force: bool) -> None:
|
|
"""Delete a snapshot."""
|
|
snapshot_dir = ctx.obj["snapshot_dir"]
|
|
manager = SnapshotManager(snapshot_dir)
|
|
|
|
if not force:
|
|
if not click.confirm(f"Delete snapshot '{name}'?"):
|
|
console.print("[yellow]Cancelled[/yellow]")
|
|
return
|
|
|
|
try:
|
|
manager.delete_snapshot(name)
|
|
console.print(f"[green]Deleted snapshot '{name}'[/green]")
|
|
except FileNotFoundError:
|
|
console.print(f"[red]Error: Snapshot '{name}' not found[/red]")
|
|
raise click.Abort()
|
|
except Exception as e:
|
|
console.print(f"[red]Error deleting snapshot: {e}[/red]")
|
|
raise click.Abort()
|