CLI Overview

# Gate your pipeline on data freshness
armor freshness check snowflake.prod.warehouse.orders || exit 1

pip install anomalyarmor-cli

# Interactive prompt
armor auth login

# Non-interactive (for CI/CD)
armor auth login --key aa_live_your_key_here

armor auth status

Authenticated as: your-email@company.com
API Key: aa_live_k8jd...9z4f
Organization: Acme Corp

armor auth logout

# List all assets
armor assets list

# Filter by source
armor assets list --source snowflake

# Filter by type
armor assets list --type table

# Get single asset
armor assets get snowflake.prod.warehouse.orders

# Get freshness summary
armor freshness summary

# List all freshness statuses
armor freshness list

# Get freshness for specific asset
armor freshness get snowflake.prod.warehouse.orders

# Check freshness (exit 0 if fresh, exit 1 if stale)
armor freshness check snowflake.prod.warehouse.orders

# Trigger refresh
armor freshness refresh snowflake.prod.warehouse.orders

# Trigger and wait for completion
armor freshness refresh snowflake.prod.warehouse.orders --wait

# Get schema drift summary
armor schema summary

# List recent schema changes
armor schema changes

# Trigger schema check
armor schema refresh snowflake.prod.warehouse.orders --wait

# List assets with lineage
armor lineage list

# Get lineage for asset
armor lineage get snowflake.prod.warehouse.orders

# Get only upstream
armor lineage get snowflake.prod.warehouse.orders --direction upstream

# Get only downstream
armor lineage get snowflake.prod.warehouse.orders --direction downstream

# Get alerts summary
armor alerts summary

# List all alerts
armor alerts list

# Filter by status
armor alerts list --status triggered

# Filter by asset
armor alerts list --asset snowflake.prod.warehouse.orders

# Get metrics summary for an asset
armor metrics summary <asset-id>

# List all metrics for an asset
armor metrics list <asset-id>

# Filter by type
armor metrics list <asset-id> --type null_percent

# Get specific metric details
armor metrics get <asset-id> <metric-id>

# Create a row count metric
armor metrics create <asset-id> --type row_count --table snowflake.prod.warehouse.orders

# Create a null percentage metric
armor metrics create <asset-id> --type null_percent --table snowflake.prod.warehouse.orders --column email

# Trigger manual capture
armor metrics capture <asset-id> <metric-id>

# Delete a metric
armor metrics delete <asset-id> <metric-id> --yes

# List your API keys
armor api-keys list

# Create new key
armor api-keys create --name "airflow-prod" --scope read-only

# Revoke key
armor api-keys revoke <key-id>

#!/bin/bash

# Gate pipeline on freshness
if armor freshness check snowflake.prod.warehouse.orders; then
    echo "Data is fresh, running pipeline..."
    dbt run
else
    echo "Data is stale, aborting"
    exit 1
fi

armor assets list

QUALIFIED NAME                        TYPE    SOURCE     ROWS
snowflake.prod.warehouse.orders       table   snowflake  1.5M
snowflake.prod.warehouse.customers    table   snowflake  250K
snowflake.prod.warehouse.products     table   snowflake  10K

armor assets list --format json

[
  {
    "qualified_name": "snowflake.prod.warehouse.orders",
    "asset_type": "table",
    "source": "snowflake",
    "row_count": 1500000
  }
]

api_key: aa_live_your_key_here
api_url: https://api.anomalyarmor.ai

export ARMOR_API_KEY="aa_live_xxx"
export ARMOR_API_URL="https://api.anomalyarmor.ai"

- name: Check data freshness
  env:
    ARMOR_API_KEY: ${{ secrets.ARMOR_API_KEY }}
  run: |
    pip install anomalyarmor-cli
    armor freshness check snowflake.prod.warehouse.orders

check_freshness:
  script:
    - pip install anomalyarmor-cli
    - armor freshness check snowflake.prod.warehouse.orders
  variables:
    ARMOR_API_KEY: $ARMOR_API_KEY