Validity API - AnomalyArmor

The Validity API enables programmatic management of data validity rules. Use it to enforce data quality constraints, detect invalid records, and integrate validation into your pipelines.

Endpoints

Method	Endpoint	Description
GET	`/api/v1/sdk/validity/{asset_id}/summary`	Get validity summary for an asset
GET	`/api/v1/sdk/validity/{asset_id}`	List validity rules for an asset
GET	`/api/v1/sdk/validity/{asset_id}/{rule_id}`	Get validity rule details
POST	`/api/v1/sdk/validity/{asset_id}`	Create a new validity rule
PATCH	`/api/v1/sdk/validity/{asset_id}/{rule_id}`	Update a validity rule
DELETE	`/api/v1/sdk/validity/{asset_id}/{rule_id}`	Delete a validity rule
POST	`/api/v1/sdk/validity/{asset_id}/{rule_id}/check`	Trigger validity check
GET	`/api/v1/sdk/validity/{asset_id}/{rule_id}/results`	List check results

Rule Types

Type	Description	Configuration
`NOT_NULL`	Column must not contain null values	None
`UNIQUE`	Column values must be unique	None
`REGEX`	Values must match a regex pattern	`rule_config.pattern`
`RANGE`	Numeric values must be within range	`rule_config.min`, `rule_config.max`
`ENUM`	Values must be in allowed set	`rule_config.allowed_values`
`DATE_FORMAT`	Values must match date format	`rule_config.format`
`CUSTOM_SQL`	Custom SQL expression	`rule_config.sql_expression`

Get Validity Summary

GET /api/v1/sdk/validity/{asset_id}/summary

Returns aggregate validity statistics for an asset.

curl -H "Authorization: Bearer aa_live_xxx" \
  "https://api.anomalyarmor.ai/api/v1/sdk/validity/550e8400-e29b-41d4-a716-446655440000/summary"

from anomalyarmor import Client

client = Client()
summary = client.validity.summary("550e8400-e29b-41d4-a716-446655440000")
print(f"Total rules: {summary.total_rules}")
print(f"Passing: {summary.passing}")
print(f"Failing: {summary.failing}")

armor validity summary 550e8400-e29b-41d4-a716-446655440000

Response

{
  "data": {
    "total_rules": 12,
    "passing": 10,
    "failing": 1,
    "error": 1
  }
}

List Validity Rules

GET /api/v1/sdk/validity/{asset_id}

Query Parameters

Parameter	Type	Default	Description
`rule_type`	string	-	Filter by type (e.g., `NOT_NULL`, `REGEX`)
`is_active`	boolean	-	Filter by active status
`limit`	integer	50	Max results (max: 100)
`offset`	integer	0	Results to skip

curl -H "Authorization: Bearer aa_live_xxx" \
  "https://api.anomalyarmor.ai/api/v1/sdk/validity/550e8400-e29b-41d4-a716-446655440000?rule_type=NOT_NULL"

from anomalyarmor import Client

client = Client()

# List all rules
rules = client.validity.list("550e8400-e29b-41d4-a716-446655440000")

# Filter by type
not_null_rules = client.validity.list(
    "550e8400-e29b-41d4-a716-446655440000",
    rule_type="NOT_NULL",
)
for r in not_null_rules:
    print(f"{r.column_name}: {r.rule_type}")

armor validity list 550e8400-e29b-41d4-a716-446655440000 --type NOT_NULL

Response

{
  "data": {
    "items": [
      {
        "id": 123,
        "uuid": "v_550e8400-e29b-41d4-a716-446655440001",
        "table_path": "snowflake.prod.warehouse.orders",
        "column_name": "customer_email",
        "rule_type": "NOT_NULL",
        "name": "Customer Email Required",
        "severity": "critical",
        "is_active": true,
        "check_interval": "daily",
        "created_at": "2024-12-01T10:00:00Z"
      }
    ]
  },
  "pagination": {
    "total": 12,
    "limit": 50,
    "offset": 0,
    "has_more": false
  }
}

Get Validity Rule Details

GET /api/v1/sdk/validity/{asset_id}/{rule_id}

curl -H "Authorization: Bearer aa_live_xxx" \
  "https://api.anomalyarmor.ai/api/v1/sdk/validity/550e8400-e29b-41d4-a716-446655440000/v_550e8400-e29b-41d4-a716-446655440001"

from anomalyarmor import Client

client = Client()
rule = client.validity.get(
    "550e8400-e29b-41d4-a716-446655440000",
    "v_550e8400-e29b-41d4-a716-446655440001",
)
print(f"Rule: {rule.name}")
print(f"Type: {rule.rule_type}")
print(f"Severity: {rule.severity}")

armor validity get 550e8400-e29b-41d4-a716-446655440000 \
  v_550e8400-e29b-41d4-a716-446655440001

Response

{
  "data": {
    "id": 123,
    "uuid": "v_550e8400-e29b-41d4-a716-446655440001",
    "table_path": "snowflake.prod.warehouse.orders",
    "column_name": "customer_email",
    "rule_type": "NOT_NULL",
    "rule_config": null,
    "name": "Customer Email Required",
    "description": "Email address must not be null for valid orders",
    "severity": "critical",
    "is_active": true,
    "alert_threshold_percent": 1.0,
    "treat_null_as_valid": false,
    "check_interval": "daily"
  }
}

Create Validity Rule

POST /api/v1/sdk/validity/{asset_id}

Requires read-write or admin scope.

Request Body

Field	Type	Required	Description
`rule_type`	string	Yes	Rule type (see table above)
`table_path`	string	Yes	Full table path (catalog.schema.table)
`column_name`	string	For column rules	Column name
`rule_config`	object	For some types	Rule-specific configuration
`name`	string	No	Human-readable rule name
`description`	string	No	Rule description
`severity`	string	No	`info`, `warning`, `critical` (default: `warning`)
`error_message`	string	No	Custom error message
`alert_threshold_percent`	float	No	Alert when invalid % exceeds this (manual mode)
`detection_mode`	string	No	`manual` (fixed threshold) or `auto` (learned baseline). Default: `manual`
`sensitivity`	integer	No	Auto-mode band width, `1`-`4` standard deviations. Default: `2`
`operating_period_mode`	string	No	`off` (default), `schedule`, or `auto`. Makes the auto baseline business-hours aware.
`operating_schedule_id`	string (UUID)	For `schedule` mode	Public UUID of the operating schedule to segment by
`treat_null_as_valid`	boolean	No	Whether nulls pass (default: `false`)
`check_interval`	string	No	`hourly`, `daily`, `weekly` (default: `daily`)

Examples

curl -X POST -H "Authorization: Bearer aa_live_xxx" \
  -H "Content-Type: application/json" \
  -d '{
    "rule_type": "NOT_NULL",
    "table_path": "snowflake.prod.warehouse.orders",
    "column_name": "customer_email",
    "name": "Customer Email Required",
    "severity": "critical"
  }' \
  "https://api.anomalyarmor.ai/api/v1/sdk/validity/550e8400-e29b-41d4-a716-446655440000"

curl -X POST -H "Authorization: Bearer aa_live_xxx" \
  -H "Content-Type: application/json" \
  -d '{
    "rule_type": "REGEX",
    "table_path": "snowflake.prod.warehouse.orders",
    "column_name": "customer_email",
    "rule_config": {"pattern": "^[\\w.-]+@[\\w.-]+\\.\\w+$"},
    "name": "Valid Email Format",
    "severity": "warning"
  }' \
  "https://api.anomalyarmor.ai/api/v1/sdk/validity/550e8400-e29b-41d4-a716-446655440000"

from anomalyarmor import Client

client = Client()
asset_id = "550e8400-e29b-41d4-a716-446655440000"

# Create NOT NULL rule
not_null = client.validity.create(
    asset_id,
    rule_type="NOT_NULL",
    table_path="snowflake.prod.warehouse.orders",
    column_name="customer_email",
    name="Customer Email Required",
    severity="critical",
)

# Create REGEX rule for email validation
regex_rule = client.validity.create(
    asset_id,
    rule_type="REGEX",
    table_path="snowflake.prod.warehouse.orders",
    column_name="customer_email",
    rule_config={"pattern": r"^[\w.-]+@[\w.-]+\.\w+$"},
    name="Valid Email Format",
    severity="warning",
)

# Create RANGE rule for numeric values
range_rule = client.validity.create(
    asset_id,
    rule_type="RANGE",
    table_path="snowflake.prod.warehouse.orders",
    column_name="order_total",
    rule_config={"min": 0, "max": 1000000},
    name="Order Total Valid Range",
)

# Create ENUM rule for status values
enum_rule = client.validity.create(
    asset_id,
    rule_type="ENUM",
    table_path="snowflake.prod.warehouse.orders",
    column_name="status",
    rule_config={"allowed_values": ["pending", "processing", "shipped", "delivered"]},
    name="Valid Order Status",
)

# Create an auto-baseline rule (no hand-set threshold). The rule learns the
# column's normal invalid rate and alerts only when it spikes abnormally high.
auto_rule = client.validity.create(
    asset_id,
    rule_type="REGEX",
    table_path="snowflake.prod.warehouse.orders",
    column_name="customer_email",
    rule_config={"pattern": r"^[\w.-]+@[\w.-]+\.\w+$"},
    name="Email Format (auto baseline)",
    detection_mode="auto",
    sensitivity=2,  # 1 = most sensitive, 4 = most tolerant
)

# NOT NULL rule
armor validity create 550e8400-e29b-41d4-a716-446655440000 \
  --type NOT_NULL \
  --table snowflake.prod.warehouse.orders \
  --column customer_email \
  --severity critical

# REGEX rule
armor validity create 550e8400-e29b-41d4-a716-446655440000 \
  --type REGEX \
  --table snowflake.prod.warehouse.orders \
  --column customer_email \
  --config '{"pattern": "^[\\w.-]+@[\\w.-]+\\.\\w+$"}'

Response

{
  "data": {
    "id": 124,
    "uuid": "v_550e8400-e29b-41d4-a716-446655440002",
    "table_path": "snowflake.prod.warehouse.orders",
    "column_name": "customer_email",
    "rule_type": "NOT_NULL",
    "name": "Customer Email Required",
    "severity": "critical",
    "is_active": true,
    "check_interval": "daily",
    "created_at": "2024-12-04T10:30:00Z"
  }
}

Update Validity Rule

PATCH /api/v1/sdk/validity/{asset_id}/{rule_id}

Requires read-write or admin scope.

Request Body

Field	Type	Description
`is_active`	boolean	Whether rule is active
`name`	string	Rule name
`description`	string	Rule description
`severity`	string	Severity level
`alert_threshold_percent`	float	Alert threshold (manual mode)
`detection_mode`	string	`manual` or `auto`
`sensitivity`	integer	Auto-mode band width, `1`-`4` standard deviations
`operating_period_mode`	string	`off`, `schedule`, or `auto`
`operating_schedule_id`	string (UUID)	Operating schedule to segment by (`schedule` mode)
`treat_null_as_valid`	boolean	Null handling
`check_interval`	string	Check interval

curl -X PATCH -H "Authorization: Bearer aa_live_xxx" \
  -H "Content-Type: application/json" \
  -d '{"severity": "critical", "alert_threshold_percent": 0.5}' \
  "https://api.anomalyarmor.ai/api/v1/sdk/validity/550e8400-e29b-41d4-a716-446655440000/v_550e8400-e29b-41d4-a716-446655440001"

from anomalyarmor import Client

client = Client()
rule = client.validity.update(
    "550e8400-e29b-41d4-a716-446655440000",
    "v_550e8400-e29b-41d4-a716-446655440001",
    severity="critical",
    alert_threshold_percent=0.5,
)

Delete Validity Rule

DELETE /api/v1/sdk/validity/{asset_id}/{rule_id}

Requires read-write or admin scope.

curl -X DELETE -H "Authorization: Bearer aa_live_xxx" \
  "https://api.anomalyarmor.ai/api/v1/sdk/validity/550e8400-e29b-41d4-a716-446655440000/v_550e8400-e29b-41d4-a716-446655440001"

from anomalyarmor import Client

client = Client()
client.validity.delete(
    "550e8400-e29b-41d4-a716-446655440000",
    "v_550e8400-e29b-41d4-a716-446655440001",
)

armor validity delete 550e8400-e29b-41d4-a716-446655440000 \
  v_550e8400-e29b-41d4-a716-446655440001 --yes

Trigger Validity Check

POST /api/v1/sdk/validity/{asset_id}/{rule_id}/check

Requires read-write or admin scope.

Request Body

Field	Type	Default	Description
`sample_limit`	integer	10	Max invalid samples to collect

curl -X POST -H "Authorization: Bearer aa_live_xxx" \
  -H "Content-Type: application/json" \
  -d '{"sample_limit": 20}' \
  "https://api.anomalyarmor.ai/api/v1/sdk/validity/550e8400-e29b-41d4-a716-446655440000/v_550e8400-e29b-41d4-a716-446655440001/check"

from anomalyarmor import Client

client = Client()
result = client.validity.check(
    "550e8400-e29b-41d4-a716-446655440000",
    "v_550e8400-e29b-41d4-a716-446655440001",
    sample_limit=20,
)

if result.status == "fail":
    print(f"Found {result.invalid_count} invalid records ({result.invalid_percent:.2f}%)")
    for sample in (result.invalid_samples or {}).get("samples", []):
        print(f"  Invalid: {sample}")
else:
    print("Validation passed!")

armor validity check 550e8400-e29b-41d4-a716-446655440000 \
  v_550e8400-e29b-41d4-a716-446655440001

Response

{
  "data": {
    "id": 456,
    "validity_rule_id": 123,
    "status": "fail",
    "total_rows": 10000,
    "invalid_count": 25,
    "invalid_percent": 0.25,
    "invalid_samples": {
      "samples": [
        {"row_id": 1001, "value": null},
        {"row_id": 1042, "value": null}
      ]
    },
    "execution_duration_ms": 1250,
    "checked_at": "2024-12-04T10:35:00Z"
  }
}

List Check Results

GET /api/v1/sdk/validity/{asset_id}/{rule_id}/results

Query Parameters

Parameter	Type	Default	Description
`limit`	integer	100	Max results
`offset`	integer	0	Results to skip

curl -H "Authorization: Bearer aa_live_xxx" \
  "https://api.anomalyarmor.ai/api/v1/sdk/validity/550e8400-e29b-41d4-a716-446655440000/v_550e8400-e29b-41d4-a716-446655440001/results?limit=30"

from anomalyarmor import Client

client = Client()
results = client.validity.results(
    "550e8400-e29b-41d4-a716-446655440000",
    "v_550e8400-e29b-41d4-a716-446655440001",
    limit=30,
)
for r in results:
    print(f"{r.checked_at}: {r.status} ({r.invalid_count} invalid)")

Response

{
  "data": {
    "items": [
      {
        "id": 456,
        "validity_rule_id": 123,
        "status": "fail",
        "total_rows": 10000,
        "invalid_count": 25,
        "invalid_percent": 0.25,
        "execution_duration_ms": 1250,
        "checked_at": "2024-12-04T10:35:00Z"
      },
      {
        "id": 455,
        "validity_rule_id": 123,
        "status": "pass",
        "total_rows": 9975,
        "invalid_count": 0,
        "invalid_percent": 0.0,
        "execution_duration_ms": 1100,
        "checked_at": "2024-12-03T10:35:00Z"
      }
    ]
  },
  "pagination": {
    "total": 60,
    "limit": 30,
    "offset": 0,
    "has_more": true
  }
}

Use Case: Validate Email Format

Ensure all customer emails match a valid format:

from anomalyarmor import Client

client = Client()
asset_id = "550e8400-e29b-41d4-a716-446655440000"

# Create email validation rule
rule = client.validity.create(
    asset_id,
    rule_type="REGEX",
    table_path="snowflake.prod.warehouse.customers",
    column_name="email",
    rule_config={"pattern": r"^[\w.-]+@[\w.-]+\.\w{2,}$"},
    name="Valid Email Format",
    description="Validates email addresses match standard format",
    severity="warning",
    alert_threshold_percent=1.0,  # Alert if > 1% invalid
)

# Run initial check
result = client.validity.check(asset_id, rule.uuid, sample_limit=20)

if result.status == "fail":
    print(f"Warning: {result.invalid_count} invalid emails found")
    print(f"Invalid rate: {result.invalid_percent:.2f}%")
    for sample in (result.invalid_samples or {}).get("samples", [])[:5]:
        print(f"  - {sample}")
else:
    print("All emails are valid!")

Error Responses

Rule Not Found (404)

{
  "error": {
    "code": "RULE_NOT_FOUND",
    "message": "Validity rule not found",
    "details": {"rule_id": "v_invalid-uuid"}
  }
}

Invalid Rule Configuration (400)

{
  "error": {
    "code": "VALIDATION_ERROR",
    "message": "Invalid regex pattern in rule_config",
    "details": {"field": "rule_config.pattern", "error": "Invalid regex syntax"}
  }
}

Forbidden (403)

{
  "error": {
    "code": "FORBIDDEN",
    "message": "Insufficient permissions. Required scope: read-write",
    "details": {"current_scope": "read-only", "required_scope": "read-write"}
  }
}

Common Questions

Which validity rule type should I pick?

Use NOT_NULL for required columns, UNIQUE for primary-key-like invariants, REGEX for string formats (emails, IDs), RANGE for numeric bounds, ENUM for finite allowed sets, DATE_FORMAT for date strings, and CUSTOM_SQL when no built-in type fits. Each type’s required rule_config is listed in the Rule Types table at the top of this page.

How do I inspect rows that failed a validity rule?

Every check response includes invalid_samples.samples, up to sample_limit rows (default 10, configurable per check). Bump sample_limit up to 100 when debugging a broken ingest. The response also returns invalid_count and invalid_percent so you can report a failure rate even when individual samples aren’t needed.

What does the alert_threshold_percent field control?

alert_threshold_percent is the invalid-row percentage that flips a check result from pass to fail in manual mode. Set it to 0 if any single invalid row should page you. Use higher values (e.g. 1.0) on rules where a small amount of invalid data is tolerated and you only want to catch systemic regressions. It is ignored in auto mode.

When should I use detection_mode `auto` instead of a fixed threshold?

Use auto when you don’t know the right threshold up front, or when a column always carries some baseline level of invalid values that you don’t want to alert on. In auto mode the rule learns the column’s historical invalid_percent distribution (with weekly seasonality once enough history accrues) and alerts only when the current check is an anomalously high deviation from that baseline. It needs at least 7 prior check results before it starts evaluating; until then it records results without alerting (a cold-start period). Lower sensitivity values (toward 1) flag smaller deviations and alert more often; higher values (toward 4) tolerate wider swings. A normal, in-baseline invalid rate stays pass even when invalid_count is greater than zero. Each auto-mode result also returns the learned band as expected_value, expected_range_low, and expected_range_high.

What does operating_period_mode do?

operating_period_mode makes the auto baseline business-hours aware so closed-period quiet does not contaminate the open-period band. With off (the default) the rule pools all prior checks into one baseline. With schedule it segments the baseline using a linked operating schedule (declared days, hours, and timezone) referenced by operating_schedule_id. With auto it learns an active/dormant calendar from the rule’s own history. An active-period check is compared only against prior active checks; a dormant-period check never alerts on a low invalid rate and alerts only on unexpected activity above a near-zero floor. It composes with detection_mode="auto" and, like the baseline, takes effect once enough history accrues. Get the schedule UUID from the operating-schedules UI or the /operating-schedules API.

How is treat_null_as_valid different from using NOT_NULL?

treat_null_as_valid governs how a non-NULL-type rule (REGEX, RANGE, ENUM, etc.) handles NULLs. When true, NULL rows are skipped; when false, NULLs count as invalid. Combine with a separate NOT_NULL rule when you need to enforce both non-null and format at once - they surface as two distinct checks you can alert on independently.

​Endpoints

​Rule Types

​Get Validity Summary

​Response

​List Validity Rules

​Query Parameters

​Response

​Get Validity Rule Details

​Response

​Create Validity Rule

​Request Body

​Examples

​Response

​Update Validity Rule

​Request Body

​Delete Validity Rule

​Trigger Validity Check

​Request Body

​Response

​List Check Results

​Query Parameters

​Response

​Use Case: Validate Email Format

​Error Responses

​Rule Not Found (404)

​Invalid Rule Configuration (400)

​Forbidden (403)

​Common Questions

​Which validity rule type should I pick?

​How do I inspect rows that failed a validity rule?

​What does the alert_threshold_percent field control?

​When should I use detection_mode auto instead of a fixed threshold?

​What does operating_period_mode do?

​How is treat_null_as_valid different from using NOT_NULL?

Endpoints

Rule Types

Get Validity Summary

Response

List Validity Rules

Query Parameters

Response

Get Validity Rule Details

Response

Create Validity Rule

Request Body

Examples

Response

Update Validity Rule

Request Body

Delete Validity Rule

Trigger Validity Check

Request Body

Response

List Check Results

Query Parameters

Response

Use Case: Validate Email Format

Error Responses

Rule Not Found (404)

Invalid Rule Configuration (400)

Forbidden (403)

Common Questions

Which validity rule type should I pick?

How do I inspect rows that failed a validity rule?

What does the alert_threshold_percent field control?

When should I use detection_mode `auto` instead of a fixed threshold?

What does operating_period_mode do?

How is treat_null_as_valid different from using NOT_NULL?