> ## Documentation Index > Fetch the complete documentation index at: https://docs.anomalyarmor.ai/llms.txt > Use this file to discover all available pages before exploring further. # First API Integration > Step-by-step guide to your first programmatic integration with AnomalyArmor 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 ```bash theme={null} 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)** ```bash theme={null} export ARMOR_API_KEY="aa_live_your_key_here" ``` **Option B: Config file (for local development)** ```bash theme={null} armor auth login # Follow the prompts to enter your API key ``` ## Step 3: Verify Connection Test that everything is working: ```python theme={null} 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: ```python theme={null} 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: ```python theme={null} 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: ```python theme={null} 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: ```bash theme={null} #!/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: ```python theme={null} # 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: ```python theme={null} # 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: ```python theme={null} # 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](https://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). Add quality gates to dbt workflows Run checks in CI/CD pipelines Integrate with Apache Airflow Full API documentation