CLI Reference

The aptdata command-line interface emits structured JSON on every outcome, making it suitable for use inside AI orchestrators, CI/CD pipelines and shell scripts.

---

aptdata run

Run a registered pipeline by name.

aptdata run PIPELINE [OPTIONS]

Arguments

Name	Required	Description
PIPELINE	✅	Pipeline identifier registered in the plugin registry

Options

Flag	Default	Description
--env, -e	dev	Target execution environment label (e.g. dev, staging, prod)
--dry-run	false	Compile and validate the pipeline without executing run()
--help		Show help and exit

Exit codes

Code	Meaning
0	Pipeline completed successfully
1	An error occurred (pipeline not found, runtime exception, etc.)

JSON events

pipeline.started – emitted immediately after the CLI receives the command:

{
  "event": "pipeline.started",
  "pipeline": "my_pipeline",
  "env": "prod",
  "dry_run": false
}

pipeline.completed – emitted when the pipeline finishes successfully:

{
  "event": "pipeline.completed",
  "pipeline": "my_pipeline",
  "env": "prod",
  "dry_run": false,
  "elapsed_seconds": 1.234
}

pipeline.error – emitted to stderr when an error occurs:

{
  "event": "pipeline.error",
  "pipeline": "my_pipeline",
  "env": "prod",
  "error": "Pipeline 'my_pipeline' not found in registry.",
  "elapsed_seconds": 0.001
}

Examples

# Run in the default dev environment
aptdata run my_pipeline

# Run against production
aptdata run my_pipeline --env prod

# Validate without executing
aptdata run my_pipeline --dry-run

# Capture and parse JSON output with jq
aptdata run my_pipeline | jq '.elapsed_seconds'

---

aptdata monitor

Launch the interactive TUI monitoring dashboard.

aptdata monitor [OPTIONS]

Options

Flag	Default	Description
--refresh, -r	1.0	Dashboard auto-refresh interval in seconds
--help		Show help and exit

Key bindings

Key	Action
r	Manually refresh all panels
q / Ctrl+C	Quit

Examples

# Open with default 1-second refresh
aptdata monitor

# Faster refresh for high-frequency pipelines
aptdata monitor --refresh 0.25

---

---

aptdata system

Inspect and validate registered systems.

aptdata system list [--json]

List all systems in the plugin registry.

aptdata system list
aptdata system list --json

aptdata system info NAME [--json]

Show detailed info about a registered system (class name, module, docstring).

aptdata system info my_pipeline
aptdata system info my_pipeline --json

aptdata system validate NAME

Instantiate the system and compile all its flows without executing.

aptdata system validate my_pipeline

---

aptdata plugin

Manage and inspect registered reader / writer plugins.

aptdata plugin list [--json]

List all registered readers and writers.

aptdata plugin list
aptdata plugin list --json

aptdata plugin inspect NAME [--json]

Show constructor argument schema for a plugin.

aptdata plugin inspect csv_reader
aptdata plugin inspect csv_reader --json

aptdata plugin preview READER [--limit N]

Execute a reader and display the first N records (default: 5).

aptdata plugin preview csv_reader --limit 10

aptdata plugin load MODULE_PATH

Dynamically import a Python module (for plugin discovery).

aptdata plugin load my_package.plugins

---

aptdata config

Manage declarative YAML pipeline configurations.

aptdata config validate PATH

Parse and validate a YAML config file.

aptdata config validate pipeline.yaml

aptdata config init [--output PATH]

Generate a starter YAML configuration template.

aptdata config init
aptdata config init --output my_pipeline.yaml

aptdata config show PATH

Pretty-print a YAML config file with syntax highlighting.

aptdata config show pipeline.yaml

aptdata config run PATH [--env ENV]

Parse a YAML config, register the system, and execute it.

aptdata config run pipeline.yaml
aptdata config run pipeline.yaml --env prod

---

aptdata telemetry

Inspect OpenTelemetry telemetry configuration.

aptdata telemetry status [--json]

Show whether OpenTelemetry is configured and the active tracer provider.

aptdata telemetry status
aptdata telemetry status --json

aptdata telemetry export [--format json]

Export collected telemetry spans/metrics as JSON.

aptdata telemetry export
aptdata telemetry export --format json

---

aptdata interactive

Launch the guided interactive wizard.

aptdata interactive

See CLI Interactive Wizard for full documentation.

---

App module

aptdata.cli.app

Typer-based static CLI for aptdata.

Design goals

• Machine / AI-readable: every outcome is emitted as a single JSON line on stdout (success) or stderr (error).
• Exit codes: 0 = success, 1 = error.
• Self-documenting: Typer generates --help automatically from the docstrings and type annotations.

Functions

run(pipeline=typer.Argument(..., help='Pipeline name / identifier to run.'), env=typer.Option('dev', '--env', '-e', help='Target execution environment.'), dry_run=typer.Option(False, '--dry-run', help='Validate and compile the pipeline without executing it.'))

Run a registered data pipeline.

Emits structured JSON logs and returns exit code 0 on success or 1 on failure so that orchestrators and AI agents can parse the outcome.

Examples:

aptdata run pipeline_x --env prod aptdata run pipeline_x --env staging --dry-run

Source code in aptdata/cli/app.py

@app.command()
def run(
    pipeline: str = typer.Argument(..., help="Pipeline name / identifier to run."),
    env: str = typer.Option("dev", "--env", "-e", help="Target execution environment."),
    dry_run: bool = typer.Option(
        False,
        "--dry-run",
        help="Validate and compile the pipeline without executing it.",
    ),
) -> None:
    """Run a registered data pipeline.

    Emits structured JSON logs and returns exit code 0 on success or 1 on
    failure so that orchestrators and AI agents can parse the outcome.

    Examples
    --------
    aptdata run pipeline_x --env prod
    aptdata run pipeline_x --env staging --dry-run
    """
    started_at = time.time()
    _emit(
        {
            "event": "pipeline.started",
            "pipeline": pipeline,
            "env": env,
            "dry_run": dry_run,
        }
    )

    try:
        # Plugin registry look-up (stub – real implementations are in plugins/)
        from aptdata.plugins import registry  # noqa: PLC0415

        pipeline_cls = registry.get(pipeline)
        if pipeline_cls is None:
            raise LookupError(f"Pipeline '{pipeline}' not found in registry.")

        instance = pipeline_cls(system_id=pipeline)

        if not dry_run:
            instance.run()

        elapsed = round(time.time() - started_at, 3)
        _emit(
            {
                "event": "pipeline.completed",
                "pipeline": pipeline,
                "env": env,
                "dry_run": dry_run,
                "elapsed_seconds": elapsed,
            }
        )
        raise SystemExit(0)

    except LookupError as exc:
        elapsed = round(time.time() - started_at, 3)
        _emit(
            {
                "event": "pipeline.error",
                "pipeline": pipeline,
                "env": env,
                "error": str(exc),
                "elapsed_seconds": elapsed,
            },
            error=True,
        )
        raise SystemExit(1) from exc

    except Exception as exc:  # noqa: BLE001
        elapsed = round(time.time() - started_at, 3)
        _emit(
            {
                "event": "pipeline.error",
                "pipeline": pipeline,
                "env": env,
                "error": str(exc),
                "elapsed_seconds": elapsed,
            },
            error=True,
        )
        raise SystemExit(1) from exc

monitor(refresh=typer.Option(1.0, '--refresh', '-r', help='Dashboard refresh interval in seconds.'))

Open the interactive TUI monitoring dashboard.

Displays the pipeline DAG, memory usage and task status in real time. Press q or Ctrl+C to exit.

Examples:

aptdata monitor aptdata monitor --refresh 0.5

Source code in aptdata/cli/app.py

@app.command()
def monitor(
    refresh: float = typer.Option(
        1.0,
        "--refresh",
        "-r",
        help="Dashboard refresh interval in seconds.",
    ),
) -> None:
    """Open the interactive TUI monitoring dashboard.

    Displays the pipeline DAG, memory usage and task status in real time.
    Press **q** or **Ctrl+C** to exit.

    Examples
    --------
    aptdata monitor
    aptdata monitor --refresh 0.5
    """
    from aptdata.tui.monitor import MonitorApp  # noqa: PLC0415

    app_instance = MonitorApp(refresh_interval=refresh)
    app_instance.run()