Rule-based Routing¶

실무 운영 가이드에서 Rule-based, Routing, Python, Jinja2을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.

개요¶

CheckpointResult
       │
       ▼
┌──────────────────┐
│   ActionRouter   │
│  (RouteMode)     │
└────────┬─────────┘
         │
    ┌────┴────┐
    ▼         ▼
┌───────┐ ┌───────┐
│Route 1│ │Route 2│ ...
│(Rule) │ │(Rule) │
└───┬───┘ └───┬───┘
    │         │
    ▼         ▼
[Actions]  [Actions]

Core Classes¶

RouteContext¶

실무 운영 가이드에서 Context을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.

@dataclass(frozen=True)
class RouteContext:
    """Routing context (immutable)."""
    checkpoint_name: str           # Checkpoint name
    run_id: str                    # Execution ID
    status: str                    # Result status
    data_asset: str                # Data asset
    run_time: datetime             # Execution time
    total_issues: int = 0          # Total issue count
    critical_issues: int = 0       # Critical issue count
    high_issues: int = 0           # High issue count
    medium_issues: int = 0         # Medium issue count
    low_issues: int = 0            # Low issue count
    info_issues: int = 0           # Info issue count
    pass_rate: float = 100.0       # Pass rate (0-100)
    tags: dict[str, str] = field(default_factory=dict)
    metadata: dict[str, Any] = field(default_factory=dict)
    validation_duration_ms: float = 0.0  # Validation duration (ms)
    error: str | None = None       # Error message

ActionRouter¶

실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.

from truthound.checkpoint.routing import ActionRouter, Route
from truthound.checkpoint.routing.base import RouteMode

class RouteMode(str, Enum):
    """Routing mode."""
    FIRST_MATCH = "first_match"      # Execute only the first matching route
    ALL_MATCHES = "all_matches"      # Execute all matching routes
    PRIORITY_GROUP = "priority_group" # Execute the highest priority group

# Create Router
router = ActionRouter(mode=RouteMode.ALL_MATCHES)

# Add Route
router.add_route(route)

Route¶

실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.

from truthound.checkpoint.routing import Route
from truthound.checkpoint.routing.base import RoutePriority

class RoutePriority(int, Enum):
    """Route priority."""
    CRITICAL = 100
    HIGH = 80
    NORMAL = 50
    LOW = 20
    DEFAULT = 0

route = Route(
    name="critical_alerts",
    rule=SeverityRule(min_severity="critical"),
    actions=[PagerDutyAction(...)],
    priority=RoutePriority.CRITICAL,  # Or an integer value
)

실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.

Built-in Rules (11 Types)¶

AlwaysRule / NeverRule¶

실무 운영 가이드에서 Rules을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.

from truthound.checkpoint.routing.rules import AlwaysRule, NeverRule

# For default route
always = AlwaysRule()  # Always True

# For disabling a route
never = NeverRule()  # Always False

SeverityRule¶

실무 운영 가이드에서 Matches을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.

from truthound.checkpoint.routing.rules import SeverityRule

# Match if critical issues exist
rule = SeverityRule(min_severity="critical")

# Match if 5 or more high-level issues exist
rule = SeverityRule(min_severity="high", min_count=5)

# Medium issues only (range specification)
rule = SeverityRule(min_severity="medium", max_severity="medium")

# Exactly 3 critical issues
rule = SeverityRule(min_severity="critical", exact_count=3)

실무 운영 가이드에서 critical, high, medium, low, info, Severity을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.

IssueCountRule¶

실무 운영 가이드에서 Matches을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.

from truthound.checkpoint.routing.rules import IssueCountRule

# 10 or more issues
rule = IssueCountRule(min_issues=10)

# Between 5 and 20 issues
rule = IssueCountRule(min_issues=5, max_issues=20)

# 3 or more critical issues
rule = IssueCountRule(min_issues=3, count_type="critical")

실무 운영 가이드에서 total, critical, high, medium, low, info을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.

StatusRule¶

Matches based on 체크포인트 status.

from truthound.checkpoint.routing.rules import StatusRule

# Failure or error
rule = StatusRule(statuses=["failure", "error"])

# All statuses except success
rule = StatusRule(statuses=["success"], negate=True)

TagRule¶

실무 운영 가이드에서 Matches을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.

from truthound.checkpoint.routing.rules import TagRule

# env=prod tag
rule = TagRule(tags={"env": "prod"})

# env=prod AND team=data
rule = TagRule(tags={"env": "prod", "team": "data"}, match_all=True)

# env=prod OR team=data
rule = TagRule(tags={"env": "prod", "team": "data"}, match_all=False)

# Check if 'critical' tag exists (value irrelevant)
rule = TagRule(tags={"critical": None})

# Match if tag is absent
rule = TagRule(tags={"env": "prod"}, negate=True)

DataAssetRule¶

실무 운영 가이드에서 Matches을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.

from truthound.checkpoint.routing.rules import DataAssetRule

# Glob pattern
rule = DataAssetRule(pattern="sales_*")

# Regular expression
rule = DataAssetRule(pattern=r"^prod_.*_v\d+$", is_regex=True)

# Case insensitive
rule = DataAssetRule(pattern="USERS*", case_sensitive=False)

MetadataRule¶

실무 운영 가이드에서 Matches을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.

from truthound.checkpoint.routing.rules import MetadataRule

# Simple comparison
rule = MetadataRule(key_path="region", expected_value="us-east-1")

# Nested path
rule = MetadataRule(key_path="config.settings.mode", expected_value="production")

# Comparison operators
rule = MetadataRule(key_path="priority", expected_value=5, comparator="gt")  # > 5
rule = MetadataRule(key_path="priority", expected_value=5, comparator="gte")  # >= 5
rule = MetadataRule(key_path="priority", expected_value=5, comparator="lt")   # < 5

# Contains check
rule = MetadataRule(key_path="owners", expected_value="data-team", comparator="contains")

# Regular expression
rule = MetadataRule(key_path="name", expected_value=r"v\d+", comparator="regex")

# Existence check
rule = MetadataRule(key_path="special_flag", comparator="exists")

실무 운영 가이드에서 eq, ne, gt, gte, lt, lte, contains, regex을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.

TimeWindowRule¶

실무 운영 가이드에서 Matches, Useful을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.

from truthound.checkpoint.routing.rules import TimeWindowRule

# Business hours (9 AM - 5 PM, weekdays)
rule = TimeWindowRule(
    start_time="09:00",
    end_time="17:00",
    days_of_week=[0, 1, 2, 3, 4],  # Mon-Fri
)

# Non-business hours (crossing midnight)
rule = TimeWindowRule(
    start_time="17:00",
    end_time="09:00",  # Until 9 AM the next day
)

# Weekends
rule = TimeWindowRule(
    days_of_week=[5, 6],  # Sat-Sun
)

# Specify timezone
rule = TimeWindowRule(
    start_time="09:00",
    end_time="17:00",
    timezone="America/New_York",
)

PassRateRule¶

실무 운영 가이드에서 Matches을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.

from truthound.checkpoint.routing.rules import PassRateRule

# Pass rate below 90%
rule = PassRateRule(max_rate=90.0)

# Pass rate between 50-80%
rule = PassRateRule(min_rate=50.0, max_rate=80.0)

# Pass rate 95% or above
rule = PassRateRule(min_rate=95.0)

ErrorRule¶

실무 운영 가이드에서 Matches을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.

from truthound.checkpoint.routing.rules import ErrorRule

# If error exists
rule = ErrorRule()

# Timeout error
rule = ErrorRule(pattern=r"timeout|timed out")

# If no error
rule = ErrorRule(negate=True)

실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.

Combinators¶

AllOf (AND)¶

실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.

from truthound.checkpoint.routing import AllOf
from truthound.checkpoint.routing.rules import SeverityRule, TagRule

# Critical issues AND production environment
rule = AllOf([
    SeverityRule(min_severity="critical"),
    TagRule(tags={"env": "prod"}),
])

AnyOf (OR)¶

실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.

from truthound.checkpoint.routing import AnyOf

# Critical issues OR error occurred
rule = AnyOf([
    SeverityRule(min_severity="critical"),
    ErrorRule(),
])

NotRule (NOT)¶

Inverts the rule 결과.

from truthound.checkpoint.routing import NotRule

# Not production environment
rule = NotRule(TagRule(tags={"env": "prod"}))

Complex Combinations¶

# (Critical OR Error) AND Production AND Business hours
complex_rule = AllOf([
    AnyOf([
        SeverityRule(min_severity="critical"),
        ErrorRule(),
    ]),
    TagRule(tags={"env": "prod"}),
    TimeWindowRule(start_time="09:00", end_time="18:00"),
])

실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.

Routing Modes¶

FIRST_MATCH¶

실무 운영 가이드에서 Executes을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.

router = ActionRouter(mode=RouteMode.FIRST_MATCH)

# Evaluated in priority order
router.add_route(Route(
    name="critical",
    rule=SeverityRule(min_severity="critical"),
    actions=[PagerDutyAction(...)],
    priority=100,  # Evaluated first
))

router.add_route(Route(
    name="high",
    rule=SeverityRule(min_severity="high"),
    actions=[SlackNotification(...)],
    priority=80,  # Evaluated next
))

# If critical issues exist, only PagerDuty is called

ALL_MATCHES¶

실무 운영 가이드에서 Executes을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.

router = ActionRouter(mode=RouteMode.ALL_MATCHES)

# Execute all matching routes
router.add_route(Route(
    name="always_store",
    rule=AlwaysRule(),
    actions=[StoreValidationResult(...)],
))

router.add_route(Route(
    name="failure_alert",
    rule=StatusRule(statuses=["failure"]),
    actions=[SlackNotification(...)],
))

# On failure: Both StoreValidationResult + SlackNotification are executed

PRIORITY_GROUP¶

실무 운영 가이드에서 Executes을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.

router = ActionRouter(mode=RouteMode.PRIORITY_GROUP)

router.add_route(Route(
    name="critical_pagerduty",
    rule=SeverityRule(min_severity="critical"),
    actions=[PagerDutyAction(...)],
    priority=100,
))

router.add_route(Route(
    name="critical_slack",
    rule=SeverityRule(min_severity="critical"),
    actions=[SlackNotification(...)],
    priority=100,  # Same priority
))

router.add_route(Route(
    name="high_slack",
    rule=SeverityRule(min_severity="high"),
    actions=[SlackNotification(...)],
    priority=80,
))

# Critical issues: Both routes with priority=100 are executed
# High issues only: Only the route with priority=80 is executed

실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.

Complete Example¶

from truthound.checkpoint import Checkpoint
from truthound.checkpoint.routing import (
    ActionRouter, Route, AllOf, AnyOf
)
from truthound.checkpoint.routing.base import RouteMode
from truthound.checkpoint.routing.rules import (
    SeverityRule, TagRule, TimeWindowRule, StatusRule, AlwaysRule
)
from truthound.checkpoint.actions import (
    StoreValidationResult, SlackNotification, PagerDutyAction, EmailNotification
)

# Create Router
router = ActionRouter(mode=RouteMode.ALL_MATCHES)

# 1. Always store results
router.add_route(Route(
    name="always_store",
    rule=AlwaysRule(),
    actions=[StoreValidationResult(store_path="./results")],
    priority=0,
))

# 2. Critical + Production → PagerDuty
router.add_route(Route(
    name="critical_prod",
    rule=AllOf([
        SeverityRule(min_severity="critical"),
        TagRule(tags={"env": "prod"}),
    ]),
    actions=[
        PagerDutyAction(routing_key="${PAGERDUTY_KEY}"),
        SlackNotification(
            webhook_url="${SLACK_WEBHOOK}",
            mention_on_failure=["@oncall"],
        ),
    ],
    priority=100,
))

# 3. Critical outside business hours → Email only
router.add_route(Route(
    name="critical_offhours",
    rule=AllOf([
        SeverityRule(min_severity="critical"),
        TimeWindowRule(start_time="18:00", end_time="09:00"),  # Non-business hours
    ]),
    actions=[EmailNotification(
        to_addresses=["oncall@example.com"],
    )],
    priority=90,
))

# 4. High issues → Slack
router.add_route(Route(
    name="high_alert",
    rule=SeverityRule(min_severity="high"),
    actions=[SlackNotification(
        webhook_url="${SLACK_WEBHOOK}",
        channel="#data-quality",
    )],
    priority=80,
))

# Connect Router to Checkpoint
checkpoint = Checkpoint(
    name="production_check",
    data_source="prod_data.parquet",
    validators=["null", "duplicate", "range"],
    router=router,
    tags={"env": "prod"},
)

result = checkpoint.run()

실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.

YAML 설정 (RouteConfigParser)¶

routes:
  - name: critical_prod
    priority: 100
    rule:
      type: all_of
      rules:
        - type: severity
          min_severity: critical
        - type: tag
          tags:
            env: prod
    actions:
      - type: pagerduty
        routing_key: ${PAGERDUTY_KEY}
      - type: slack
        webhook_url: ${SLACK_WEBHOOK}
        mention_on_failure:
          - "@oncall"

  - name: high_alert
    priority: 80
    rule:
      type: severity
      min_severity: high
    actions:
      - type: slack
        webhook_url: ${SLACK_WEBHOOK}
        channel: "#data-quality"

  - name: always_store
    priority: 0
    rule:
      type: always
    actions:
      - type: store_result
        store_path: ./results