First API Integration

This guide walks you through building your first programmatic integration with AnomalyArmor. By the end, you’ll have a working data quality check that can run in your pipeline.

Prerequisites

An AnomalyArmor account with at least one connected data source
Python 3.9+ installed
An API key (create in Settings > API Keys)

Step 1: Install the SDK

pip install anomalyarmor-cli

Step 2: Configure Authentication

Store your API key securely. You have two options: Option A: Environment variable (recommended for CI/CD)

export ARMOR_API_KEY="aa_live_your_key_here"

Option B: Config file (for local development)

armor auth login
# Follow the prompts to enter your API key

Step 3: Verify Connection

Test that everything is working:

from anomalyarmor import Client

client = Client()

# List your data assets
assets = client.assets.list(limit=5)
for asset in assets:
    print(f"{asset.qualified_name} ({asset.asset_type})")

You should see a list of your connected tables and views.

Step 4: Check Data Freshness

The most common integration pattern is checking data freshness before running a pipeline. Here’s a complete example:

from anomalyarmor import Client
from anomalyarmor.exceptions import StalenessError

def run_pipeline():
    client = Client()

    # Check that source data is fresh before processing
    try:
        client.freshness.require_fresh("snowflake.prod.warehouse.orders")
        print("Data is fresh, proceeding with pipeline...")
        # Your pipeline logic here
    except StalenessError as e:
        print(f"Pipeline aborted: data is {e.hours_since_update:.1f}h stale")
        raise

if __name__ == "__main__":
    run_pipeline()

Step 5: Add Data Quality Checks

Expand your integration with validity and referential integrity checks:

from anomalyarmor import Client

client = Client()
asset_id = "your-asset-uuid"  # Get from assets.list()

# Check freshness
summary = client.freshness.summary()
print(f"Overall freshness: {summary.fresh_percentage}%")

# Check validity rules
validity_summary = client.validity.summary(asset_id)
if validity_summary.failing > 0:
    print(f"Warning: {validity_summary.failing} validity rules failing")
    # List failing rules
    rules = client.validity.list(asset_id)
    for rule in rules:
        result = client.validity.check(asset_id, rule.uuid)
        if result.status == "fail":
            print(f"  - {rule.name}: {result.invalid_count} invalid records")

# Check referential integrity
ref_summary = client.referential.summary(asset_id)
if ref_summary.failing_checks > 0:
    print(f"Warning: {ref_summary.failing_checks} referential checks failing")

Step 6: Create a Quality Gate

Combine all checks into a single quality gate function:

from anomalyarmor import Client
from anomalyarmor.exceptions import StalenessError

def quality_gate(asset_id: str, critical_tables: list[str]) -> bool:
    """
    Run all data quality checks before pipeline execution.
    Returns True if all checks pass, False otherwise.

    Note: list[str] type hint requires Python 3.9+. For earlier versions,
    use typing.List[str] instead.
    """
    client = Client()
    passed = True

    # 1. Check freshness of critical tables
    for table in critical_tables:
        try:
            client.freshness.require_fresh(table)
            print(f"[PASS] Freshness: {table}")
        except StalenessError as e:
            print(f"[FAIL] Freshness: {table} ({e.hours_since_update:.1f}h stale)")
            passed = False

    # 2. Check validity rules
    validity = client.validity.summary(asset_id)
    if validity.failing == 0:
        print(f"[PASS] Validity: {validity.total_rules} rules")
    else:
        print(f"[FAIL] Validity: {validity.failing}/{validity.total_rules} rules failing")
        passed = False

    # 3. Check referential integrity
    ref = client.referential.summary(asset_id)
    if ref.failing_checks == 0:
        print(f"[PASS] Referential: {ref.total_checks} checks")
    else:
        print(f"[FAIL] Referential: {ref.failing_checks}/{ref.total_checks} failing")
        passed = False

    return passed

# Usage
if __name__ == "__main__":
    asset = "your-asset-uuid"
    tables = [
        "snowflake.prod.warehouse.orders",
        "snowflake.prod.warehouse.customers",
    ]

    if quality_gate(asset, tables):
        print("\nAll quality checks passed! Running pipeline...")
        # run_pipeline()
    else:
        print("\nQuality checks failed. Pipeline aborted.")
        exit(1)

Step 7: CLI Integration

For shell scripts and CI/CD, use the CLI directly:

#!/bin/bash
set -e

# Check freshness (exits 1 if stale)
armor freshness check snowflake.prod.warehouse.orders

# If we get here, data is fresh
echo "Data quality checks passed!"
dbt run

Common Patterns

Pattern 1: Pre-ETL Validation

Run checks before ETL starts:

# At the start of your ETL
client.freshness.require_fresh("source_table")
client.validity.check(asset_id, "not_null_rule")
# Then run ETL...

Pattern 2: Post-ETL Validation

Verify output quality after ETL:

# After ETL completes
result = client.validity.check(asset_id, "uniqueness_rule")
if result.status == "fail":
    rollback_etl()
    alert_team(f"ETL produced {result.invalid_count} duplicate records")

Pattern 3: Continuous Monitoring

Schedule regular quality checks:

# In a scheduled job (e.g., Airflow, cron)
summary = client.metrics.summary(asset_id)
if summary.failing > 0:
    alert_team(f"{summary.failing} metrics failing")

Next Steps

Common Questions

Where do I get an API key to start?

Sign in to app.anomalyarmor.ai, open Settings → API Keys, and click Create Key. New keys are shown once - copy to a password manager or secrets manager immediately. Set the scope to read-only for pipeline gating or read-write if your integration needs to create metrics or acknowledge alerts.

Should I use the Python SDK or raw REST calls for my first integration?

Python SDK if you’re in Python or Airflow - it handles pagination, retries on 429, and error typing for you. Raw REST (curl or requests) if you’re in a different language, want zero dependencies, or are prototyping a webhook handler. Both paths are first-class.

How do I test an integration without affecting production monitoring?

Create a separate AnomalyArmor workspace (or use a sandbox project) with a non-production database connection. Integrations there can create/delete monitors freely without touching prod. When ready, swap the API key to production and re-run against real endpoints.

What’s the lightest possible first integration?

A freshness check at the start of your pipeline: one API call, one conditional exit. GET /api/v1/freshness/check?asset=my_table returns whether the table meets its SLA. If no, abort the pipeline. This pattern is 5 lines of code and catches the most common data issue (stale upstream).

dbt Integration

Add quality gates to dbt workflows

GitHub Actions

Run checks in CI/CD pipelines

Airflow Integration

Integrate with Apache Airflow

API Reference

Full API documentation

API Overview

API Reference

API Guides

SDKs

CLI

Integrations

AI Agents

Prerequisites

Step 1: Install the SDK

Step 2: Configure Authentication

Step 3: Verify Connection

Step 4: Check Data Freshness

Step 5: Add Data Quality Checks

Step 6: Create a Quality Gate

Step 7: CLI Integration

Common Patterns

Pattern 1: Pre-ETL Validation

Pattern 2: Post-ETL Validation

Pattern 3: Continuous Monitoring

Next Steps

Common Questions

Where do I get an API key to start?

Should I use the Python SDK or raw REST calls for my first integration?

How do I test an integration without affecting production monitoring?

What’s the lightest possible first integration?

dbt Integration

GitHub Actions

Airflow Integration

API Reference

​Prerequisites

​Step 1: Install the SDK

​Step 2: Configure Authentication

​Step 3: Verify Connection

​Step 4: Check Data Freshness

​Step 5: Add Data Quality Checks

​Step 6: Create a Quality Gate

​Step 7: CLI Integration

​Common Patterns

​Pattern 1: Pre-ETL Validation

​Pattern 2: Post-ETL Validation

​Pattern 3: Continuous Monitoring

​Next Steps

​Common Questions

​Where do I get an API key to start?

​Should I use the Python SDK or raw REST calls for my first integration?

​How do I test an integration without affecting production monitoring?

​What’s the lightest possible first integration?

dbt Integration

GitHub Actions

Airflow Integration

API Reference

Prerequisites

Step 1: Install the SDK

Step 2: Configure Authentication

Step 3: Verify Connection

Step 4: Check Data Freshness

Step 5: Add Data Quality Checks

Step 6: Create a Quality Gate

Step 7: CLI Integration

Common Patterns

Pattern 1: Pre-ETL Validation

Pattern 2: Post-ETL Validation

Pattern 3: Continuous Monitoring

Next Steps

Common Questions

Where do I get an API key to start?

Should I use the Python SDK or raw REST calls for my first integration?

How do I test an integration without affecting production monitoring?

What’s the lightest possible first integration?