SDK Reference

from anomalyarmor import Client

client = Client(
    api_key="aa_live_xxx",  # Or use ARMOR_API_KEY env var
    api_url="https://app.anomalyarmor.ai/api/v1",  # Optional
    timeout=30,  # Request timeout in seconds
)

with Client() as client:
    assets = client.assets.list()
# Connection automatically closed

assets = client.assets.list(
    source="snowflake",      # Filter by source
    asset_type="table",      # Filter by type
    search="orders",         # Search in names
    limit=50,                # Max results (default 50, max 100)
    offset=0,                # Skip N results
)

asset = client.assets.get("snowflake.prod.warehouse.orders")
# Or by UUID
asset = client.assets.get("550e8400-e29b-41d4-a716-446655440000")

summary = client.freshness.summary()
print(f"Fresh: {summary.fresh}/{summary.total_assets}")
print(f"Fresh rate: {summary.fresh_percentage}%")

statuses = client.freshness.list(
    status="stale",     # Filter: "fresh", "stale", "unknown"
    limit=50,
    offset=0,
)

status = client.freshness.get("snowflake.prod.warehouse.orders")
print(f"Fresh: {status.is_fresh}")
print(f"Last updated: {status.last_updated}")
print(f"Hours since update: {status.hours_since_update}")

from anomalyarmor.exceptions import StalenessError

try:
    client.freshness.require_fresh(
        "snowflake.prod.warehouse.orders",
        max_age_hours=24,  # Optional custom threshold
    )
    print("Data is fresh!")
except StalenessError as e:
    print(f"Stale: {e.hours_since_update}h old")
    raise

result = client.freshness.refresh("snowflake.prod.warehouse.orders")
print(f"Job ID: {result['job_id']}")

summary = client.schema.summary()
print(f"Changes last 24h: {summary.changes_last_24h}")

changes = client.schema.changes(
    asset_id="snowflake.prod.warehouse.orders",  # Optional filter
    change_type="column_added",  # Optional filter
    limit=50,
    offset=0,
)
for change in changes:
    print(f"{change.qualified_name}: {change.change_type}")

result = client.schema.refresh("snowflake.prod.warehouse.orders")

assets = client.lineage.list(
    source="snowflake",
    has_upstream=True,
    has_downstream=True,
    limit=50,
)

lineage = client.lineage.get(
    "snowflake.prod.warehouse.orders",
    direction="both",  # "upstream", "downstream", or "both"
    depth=2,           # Levels to traverse (1-5)
)

for upstream in lineage.upstream:
    print(f"<- {upstream.qualified_name}")

for downstream in lineage.downstream:
    print(f"-> {downstream.qualified_name}")

tags = client.tags.list(
    asset="postgresql.analytics",  # Asset ID or qualified name
    category="business",           # Optional filter
)
for tag in tags:
    print(f"{tag.name} on {tag.object_path}")

tag = client.tags.create(
    asset="postgresql.analytics",
    name="pii_data",
    object_path="gold.customers",          # Required: schema.table or schema.table.column
    object_type="table",                   # "table" or "column" (default: "table")
    category="governance",                 # "business", "technical", "governance"
    description="Contains customer PII",   # Optional
)

result = client.tags.apply(
    asset="postgresql.analytics",
    tag_names=["pii", "gdpr"],                           # Required
    object_paths=["gold.customers", "gold.orders"],      # Required
    category="governance",
)
print(f"Applied: {result.applied}, Failed: {result.failed}")

answer = client.intelligence.ask(
    asset="postgresql.analytics",
    question="What tables contain customer data?",
)
print(answer.answer)
print(f"Confidence: {answer.confidence}")
print(f"Sources: {answer.sources}")

result = client.intelligence.generate(
    asset="postgresql.analytics",
)
print(f"Job ID: {result.job_id}")
print(f"Status: {result.status}")

capsule = client.investigations.explain(
    asset_id="11111111-1111-1111-1111-111111111111",
    question="Why is the orders table stale?",
    event_type="freshness",
)

print(capsule.root_cause_hypothesis)
print(f"Confidence: {capsule.confidence:.0%}")

for ev in capsule.triggers:
    citation = f"{ev.source}:{ev.source_id}" if ev.source_id else "(legacy)"
    print(f"- [{citation}] {ev.summary}")

print(capsule.suggested_fix)

capsule = client.investigations.get("aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee")
print(capsule.root_cause_hypothesis)
print(capsule.investigation_id)  # the same UUID you passed in

status = client.jobs.status("job_abc123")
print(f"Status: {status.status}")
print(f"Progress: {status.progress}%")
if status.error:
    print(f"Error: {status.error}")

summary = client.metrics.summary("asset-uuid")
print(f"Active: {summary.active_metrics}/{summary.total_metrics}")
print(f"Health: {summary.health_percentage}%")

metrics = client.metrics.list(
    "asset-uuid",
    metric_type="null_percent",  # Optional filter
    is_active=True,              # Optional filter
    limit=50,
    offset=0,
)
for m in metrics:
    print(f"{m.table_path}.{m.column_name}: {m.metric_type}")

metric = client.metrics.get(
    "asset-uuid",
    "metric-uuid",
    include_snapshots=True,
    snapshot_limit=30,
)
print(f"Type: {metric.metric_type}")
print(f"Table: {metric.table_path}")

metric = client.metrics.create(
    "asset-uuid",
    metric_type="null_percent",
    table_path="catalog.schema.table",
    column_name="email",
    capture_interval="daily",
)
print(f"Created: {metric.id}")

metric = client.metrics.update(
    "asset-uuid",
    "metric-uuid",
    is_active=False,
    sensitivity=2.0,
)

client.metrics.delete("asset-uuid", "metric-uuid")

result = client.metrics.capture("asset-uuid", "metric-uuid")
print(f"Captured {result.get('snapshot_count', 0)} snapshots")

snapshots = client.metrics.snapshots(
    "asset-uuid",
    "metric-uuid",
    limit=100,
)
for s in snapshots:
    print(f"{s.captured_at}: {s.value}")

summary = client.validity.summary("asset-uuid")
print(f"Total rules: {summary.total_rules}")
print(f"Passing: {summary.passing}, Failing: {summary.failing}")

rules = client.validity.list(
    "asset-uuid",
    rule_type="NOT_NULL",  # Optional filter
    is_active=True,        # Optional filter
    limit=50,
)
for r in rules:
    print(f"{r.column_name}: {r.rule_type}")

rule = client.validity.get("asset-uuid", "rule-uuid")
print(f"Type: {rule.rule_type}")
print(f"Severity: {rule.severity}")

# NOT_NULL rule
rule = client.validity.create(
    "asset-uuid",
    rule_type="NOT_NULL",
    table_path="catalog.schema.table",
    column_name="email",
    severity="critical",
)

# REGEX rule
regex_rule = client.validity.create(
    "asset-uuid",
    rule_type="REGEX",
    table_path="catalog.schema.table",
    column_name="email",
    rule_config={"pattern": r"^[\w.-]+@[\w.-]+\.\w+$"},
)

rule = client.validity.update(
    "asset-uuid",
    "rule-uuid",
    severity="critical",
    is_active=True,
)

client.validity.delete("asset-uuid", "rule-uuid")

result = client.validity.check(
    "asset-uuid",
    "rule-uuid",
    sample_limit=20,
)
if result.status == "fail":
    print(f"Invalid: {result.invalid_count} ({result.invalid_percent:.2f}%)")

results = client.validity.results(
    "asset-uuid",
    "rule-uuid",
    limit=30,
)
for r in results:
    print(f"{r.checked_at}: {r.status}")

summary = client.referential.summary("asset-uuid")
print(f"Total: {summary.total_checks}")
print(f"Passing: {summary.passing_checks}, Failing: {summary.failing_checks}")

checks = client.referential.list(
    "asset-uuid",
    is_active=True,
    limit=50,
)
for c in checks:
    print(f"{c.child_column_name} -> {c.parent_column_name}")

check = client.referential.get("asset-uuid", "check-uuid")
print(f"FK: {check.child_table_path}.{check.child_column_name}")
print(f"PK: {check.parent_table_path}.{check.parent_column_name}")

check = client.referential.create(
    "asset-uuid",
    child_table_path="catalog.schema.orders",
    child_column_name="customer_id",
    parent_table_path="catalog.schema.customers",
    parent_column_name="id",
    name="Orders -> Customers FK",
    max_orphan_count=0,
)

check = client.referential.update(
    "asset-uuid",
    "check-uuid",
    max_orphan_percent=0.5,
    capture_interval="hourly",
)

client.referential.delete("asset-uuid", "check-uuid")

result = client.referential.execute("asset-uuid", "check-uuid")
if result.status == "fail":
    print(f"Orphans: {result.orphan_count} ({result.orphan_percent:.2f}%)")

results = client.referential.results(
    "asset-uuid",
    "check-uuid",
    limit=30,
)
for r in results:
    print(f"{r.created_at}: {r.status} ({r.orphan_count} orphans)")

summary = client.alerts.summary()
print(f"Triggered: {summary.triggered}")
print(f"Last 24h: {summary.triggered_last_24h}")

alerts = client.alerts.list(
    status="triggered",  # "triggered", "acknowledged", "resolved"
    alert_type="freshness",  # "freshness", "schema_change", "row_count"
    asset_id="snowflake.prod.warehouse.orders",
    limit=50,
)

rules = client.alerts.rules()
for rule in rules:
    print(f"{rule.name}: {rule.alert_type}")

keys = client.api_keys.list(
    include_revoked=False,
    limit=50,
)

new_key = client.api_keys.create(
    name="Airflow Production",
    scope="read-only",  # "read-only", "read-write", "admin"
)
print(f"Key: {new_key.key}")  # Save this!

key = client.api_keys.get(key_id)
print(f"{key.name}: {key.scope}")

client.api_keys.revoke(key_id)

usage = client.api_keys.usage()
print(f"Keys: {usage['current_count']}/{usage['max_keys']}")
print(f"Rate limit: {usage['rate_limit_per_min']}/min")

class Asset:
    id: str
    qualified_name: str
    name: str
    asset_type: str  # "table", "view", "model"
    source: str
    database: str | None
    schema_name: str | None
    description: str | None
    row_count: int | None
    column_count: int | None
    tags: list[str]
    created_at: datetime
    updated_at: datetime

class FreshnessStatus:
    asset_id: str
    qualified_name: str
    status: str  # "fresh", "stale", "unknown"
    is_fresh: bool
    is_stale: bool
    last_updated: datetime | None
    hours_since_update: float | None
    staleness_threshold_hours: float | None
    checked_at: datetime

class APIKey:
    id: str
    name: str
    key_prefix: str  # e.g., "aa_live_abc1"
    key_suffix: str  # e.g., "xy9z"
    scope: str
    created_at: datetime
    last_used_at: datetime | None
    revoked_at: datetime | None

class CreatedAPIKey:
    id: str
    name: str
    key: str  # Full key - only shown once!
    scope: str
    created_at: datetime

class Tag:
    id: str
    name: str
    category: str  # "business", "technical", "governance"
    object_path: str | None  # e.g., "gold.customers"
    object_type: str | None  # "table" or "column"
    description: str | None
    created_at: datetime | None

class BulkApplyResult:
    applied: int   # Number of tags successfully applied
    failed: int    # Number that failed
    total: int     # Total attempted

class IntelligenceAnswer:
    answer: str           # AI-generated answer
    confidence: str       # "high", "medium", "low"
    sources: str          # Data sources used

class JobStatus:
    job_id: str
    status: str      # "pending", "running", "completed", "failed"
    progress: int    # 0-100
    error: str | None
    created_at: datetime
    completed_at: datetime | None

class MetricsSummary:
    total_metrics: int
    active_metrics: int
    total_checks: int
    passing: int
    failing: int
    warning: int
    error: int
    health_percentage: float

class MetricDefinition:
    id: str              # Public UUID
    internal_id: int
    asset_id: int
    table_path: str
    column_name: str | None
    metric_type: str     # row_count, null_percent, etc.
    capture_interval: str
    sensitivity: int
    is_active: bool
    created_at: datetime | None

class MetricSnapshot:
    id: int
    metric_definition_id: int
    value: float
    captured_at: datetime
    is_anomaly: bool
    z_score: float | None
    status: str | None   # PASS, FAIL, WARNING

class ValiditySummary:
    total_rules: int
    passing: int
    failing: int
    error: int

class ValidityRule:
    id: int
    uuid: str
    table_path: str
    column_name: str | None
    rule_type: str       # NOT_NULL, UNIQUE, REGEX, etc.
    rule_config: dict | None
    name: str | None
    severity: str        # info, warning, critical
    is_active: bool
    check_interval: str

class ValidityCheckResult:
    id: int
    validity_rule_id: int
    status: str          # pass, fail, error
    total_rows: int
    invalid_count: int
    invalid_percent: float
    invalid_samples: dict | None
    checked_at: datetime

class ReferentialSummary:
    total_checks: int
    active_checks: int
    passing_checks: int
    failing_checks: int
    last_check_at: datetime | None

class ReferentialCheck:
    id: str              # Public UUID
    internal_id: int
    asset_id: int
    child_table_path: str
    child_column_name: str
    parent_table_path: str
    parent_column_name: str
    name: str | None
    capture_interval: str
    max_orphan_count: int | None
    max_orphan_percent: float | None
    is_active: bool

class ReferentialCheckResult:
    id: int
    referential_check_id: int
    status: str          # pass, fail, error
    orphan_count: int
    orphan_percent: float
    total_child_rows: int
    orphan_sample: list | None
    created_at: datetime