Skip to main content
The SDK uses typed exceptions to communicate errors. All exceptions inherit from ArmorError.

​
Exception Hierarchy

Exception hierarchy showing ArmorError as base with specific error types

​
Import Exceptions

from anomalyarmor.exceptions import (
    ArmorError,           # Base exception
    AuthenticationError,  # Invalid/missing API key
    AuthorizationError,   # Valid key, insufficient scope
    NotFoundError,        # Resource not found
    ValidationError,      # Invalid parameters
    RateLimitError,       # Rate limit exceeded
    ServerError,          # Server error
    StalenessError,       # Data is stale
)

​
Exception Types

​
ArmorError

Base exception for all SDK errors. 
class ArmorError(Exception):
    message: str           # Human-readable message
    code: str | None       # Error code (e.g., "NOT_FOUND")
    details: dict          # Additional context
Example: 
try:
    client.assets.get("invalid")
except ArmorError as e:
    print(f"Error: {e.message}")
    print(f"Code: {e.code}")
    print(f"Details: {e.details}")

​
AuthenticationError

Raised when authentication fails (401). 
try:
    client = Client(api_key="invalid_key")
    client.assets.list()
except AuthenticationError as e:
    print("Invalid API key")
Common causes:
  • Invalid API key
  • Expired or revoked key
  • Missing Authorization header

​
AuthorizationError

Raised when authorization fails (403). The API key is valid but lacks permissions. 
class AuthorizationError(ArmorError):
    required_scope: str | None   # Scope needed for this action
    current_scope: str | None    # Scope of your API key
Example: 
try:
    # Trying to create a key with read-only scope
    client.api_keys.create(name="test", scope="admin")
except AuthorizationError as e:
    print(f"Need {e.required_scope}, have {e.current_scope}")
Common causes:
  • Using read-only key for write operations
  • Using read-write key for admin operations

​
NotFoundError

Raised when a resource doesn’t exist (404). 
class NotFoundError(ArmorError):
    resource_type: str | None  # e.g., "asset"
    resource_id: str | None    # The ID that wasn't found
Example: 
try:
    asset = client.assets.get("nonexistent.qualified.name")
except NotFoundError as e:
    print(f"Asset not found: {e.resource_id}")

​
ValidationError

Raised when request parameters are invalid (422). 
class ValidationError(ArmorError):
    field_errors: dict[str, str]  # Field-specific errors
Example: 
try:
    client.api_keys.create(name="", scope="invalid")
except ValidationError as e:
    print(f"Validation failed: {e.field_errors}")

​
RateLimitError

Raised when rate limit is exceeded (429). 
class RateLimitError(ArmorError):
    retry_after: int | None  # Seconds to wait before retrying
Example: 
import time

try:
    assets = client.assets.list()
except RateLimitError as e:
    if e.retry_after:
        print(f"Rate limited. Waiting {e.retry_after}s...")
        time.sleep(e.retry_after)
        # Retry

​
ServerError

Raised for server-side errors (5xx). 
class ServerError(ArmorError):
    status_code: int  # HTTP status code
Example: 
try:
    assets = client.assets.list()
except ServerError as e:
    print(f"Server error ({e.status_code}): {e.message}")

​
StalenessError

Raised by require_fresh() when data is stale. This is a data quality exception, not an API error. 
class StalenessError(ArmorError):
    asset: str                    # Asset qualified name
    hours_since_update: float     # Hours since last update
    threshold_hours: float        # The threshold that was exceeded
Example: 
from anomalyarmor.exceptions import StalenessError

try:
    client.freshness.require_fresh("snowflake.prod.warehouse.orders")
except StalenessError as e:
    print(f"Asset {e.asset} is stale")
    print(f"Last update: {e.hours_since_update:.1f}h ago")
    print(f"Threshold: {e.threshold_hours:.1f}h")
    sys.exit(1)

​
Best Practices

​
Catch Specific Exceptions

from anomalyarmor.exceptions import (
    StalenessError,
    AuthenticationError,
    RateLimitError,
    ArmorError,
)

try:
    client.freshness.require_fresh(asset)

except StalenessError as e:
    # Data quality issue - fail the pipeline
    logger.error(f"Stale data: {e.asset}")
    raise

except AuthenticationError:
    # Configuration issue - alert on-call
    logger.critical("Invalid API key!")
    notify_oncall()
    raise

except RateLimitError as e:
    # Transient - retry after waiting
    time.sleep(e.retry_after or 60)
    retry()

except ArmorError as e:
    # Unexpected error - log and continue
    logger.warning(f"API error: {e}")

​
Retry with Backoff

import time
from anomalyarmor.exceptions import RateLimitError, ServerError

def with_retry(fn, max_retries=3):
    """Execute function with exponential backoff."""
    for attempt in range(max_retries):
        try:
            return fn()

        except RateLimitError as e:
            wait = e.retry_after or (2 ** attempt * 10)
            print(f"Rate limited, waiting {wait}s...")
            time.sleep(wait)

        except ServerError as e:
            if attempt == max_retries - 1:
                raise
            wait = 2 ** attempt * 5
            print(f"Server error, retrying in {wait}s...")
            time.sleep(wait)

    raise Exception("Max retries exceeded")

# Usage
assets = with_retry(lambda: client.assets.list())

​
Pipeline Gate Pattern

from anomalyarmor import Client
from anomalyarmor.exceptions import StalenessError, ArmorError
import sys

def check_freshness_gate(assets: list[str]) -> bool:
    """Gate pipeline on data freshness."""
    client = Client()

    stale = []
    for asset in assets:
        try:
            client.freshness.require_fresh(asset)
        except StalenessError:
            stale.append(asset)
        except ArmorError as e:
            print(f"Warning: Could not check {asset}: {e}")

    if stale:
        print(f"BLOCKED: {len(stale)} stale assets: {stale}")
        return False

    print("All assets fresh, proceeding...")
    return True

# In your pipeline
if not check_freshness_gate(["orders", "customers"]):
    sys.exit(1)

​
Debugging

​
Enable Request Logging

import logging

# Enable debug logging for httpx
logging.basicConfig(level=logging.DEBUG)
logging.getLogger("httpx").setLevel(logging.DEBUG)

client = Client()

​
Inspect Error Details

try:
    client.assets.get("invalid")
except ArmorError as e:
    print(f"Message: {e.message}")
    print(f"Code: {e.code}")
    print(f"Details: {e.details}")

    # For ValidationError
    if hasattr(e, 'field_errors'):
        print(f"Field errors: {e.field_errors}")

​
Check API Key Validity

try:
    # Simple health check
    client.freshness.summary()
    print("API key is valid")
except AuthenticationError:
    print("API key is invalid or revoked")