## Import Exceptions
```python theme={null}
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.
```python theme={null}
class ArmorError(Exception):
message: str # Human-readable message
code: str | None # Error code (e.g., "NOT_FOUND")
details: dict # Additional context
```
**Example:**
```python theme={null}
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).
```python theme={null}
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.
```python theme={null}
class AuthorizationError(ArmorError):
required_scope: str | None # Scope needed for this action
current_scope: str | None # Scope of your API key
```
**Example:**
```python theme={null}
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).
```python theme={null}
class NotFoundError(ArmorError):
resource_type: str | None # e.g., "asset"
resource_id: str | None # The ID that wasn't found
```
**Example:**
```python theme={null}
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).
```python theme={null}
class ValidationError(ArmorError):
field_errors: dict[str, str] # Field-specific errors
```
**Example:**
```python theme={null}
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).
```python theme={null}
class RateLimitError(ArmorError):
retry_after: int | None # Seconds to wait before retrying
```
**Example:**
```python theme={null}
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).
```python theme={null}
class ServerError(ArmorError):
status_code: int # HTTP status code
```
**Example:**
```python theme={null}
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.
```python theme={null}
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:**
```python theme={null}
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
```python theme={null}
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
```python theme={null}
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
```python theme={null}
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
```python theme={null}
import logging
# Enable debug logging for httpx
logging.basicConfig(level=logging.DEBUG)
logging.getLogger("httpx").setLevel(logging.DEBUG)
client = Client()
```
### Inspect Error Details
```python theme={null}
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
```python theme={null}
try:
# Simple health check
client.freshness.summary()
print("API key is valid")
except AuthenticationError:
print("API key is invalid or revoked")
```
## Common Questions
### Which exception should I catch to fail a pipeline on stale data?
Catch `StalenessError`, which is raised by `client.freshness.require_fresh(asset)` when the asset is past its freshness threshold. It carries `asset`, `hours_since_update`, and `threshold_hours` so you can log actionable context. Let it propagate in Airflow tasks to mark the task failed cleanly.
### How do I distinguish a transient server error from a permanent one?
Catch `ServerError` (5xx) separately from `ArmorError` and retry with backoff; these are usually transient. `ValidationError` and `NotFoundError` are permanent for the given input, so retrying won't help. The "Retry with Backoff" example above shows the pattern for 429 + 5xx specifically.
### What's the difference between AuthenticationError and AuthorizationError?
`AuthenticationError` (401) means the API key itself is missing, invalid, or revoked. `AuthorizationError` (403) means the key is valid but lacks the scope the endpoint requires. The latter exposes `required_scope` and `current_scope` attributes so you can point users at the right key to use.
### How do I debug an unexpected error from the SDK?
Enable httpx debug logging (`logging.getLogger("httpx").setLevel(logging.DEBUG)`) to see the raw request and response, then inspect `e.message`, `e.code`, and `e.details` on the caught `ArmorError`. For `ValidationError`, `e.field_errors` points at the exact fields the API rejected.