Custom Actions¶

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

CustomAction¶

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

설정¶

실무 운영 가이드에서 Property을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 Type을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 Default을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 Description을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
실무 운영 가이드에서 `callback`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 Callable을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 None을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 `None`, None을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
실무 운영 가이드에서 `shell_command`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 None을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 `None`, None을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
실무 운영 가이드에서 `environment`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 `dict[str, str]`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 `{}`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 Environment을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
실무 운영 가이드에서 `pass_result_as_json`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 `bool`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 `True`, True을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 JSON, Pass을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
실무 운영 가이드에서 `working_directory`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 None을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 `None`, None을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
실무 운영 가이드에서 `notify_on`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 `str`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 `"always"`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	실무 운영 가이드에서 Execution을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.

Using Python Callbacks¶

from truthound.checkpoint.actions import CustomAction

def my_callback(checkpoint_result):
    """Custom logic for processing validation results."""
    status = checkpoint_result.status.value
    validation_run = checkpoint_result.validation_run
    validation = checkpoint_result.validation_view
    stats = validation.statistics if validation else None

    print(f"Checkpoint {checkpoint_result.checkpoint_name}: {status}")
    print(f"Total issues: {stats.total_issues if stats else 0}")
    print(f"Checks executed: {len(validation_run.checks) if validation_run else 0}")

    if status == "failure":
        # Custom notification logic
        send_custom_alert(checkpoint_result)

    # Save additional data
    save_to_database(checkpoint_result)

    # Return value is included in ActionResult.details
    return {"processed": True, "custom_metric": 42}


action = CustomAction(
    callback=my_callback,
    notify_on="always",
)

Async Callbacks¶

import asyncio

async def async_callback(checkpoint_result):
    """Asynchronous custom logic."""
    await asyncio.sleep(1)  # Async operation
    await send_notification_async(checkpoint_result)
    return {"async_result": True}


action = CustomAction(callback=async_callback)

Using Shell Commands¶

# Simple shell command
action = CustomAction(
    shell_command="./scripts/notify.sh",
    notify_on="failure",
)

# Pass environment variables
action = CustomAction(
    shell_command="./scripts/process_result.py",
    environment={
        "API_KEY": "${SECRET_KEY}",
        "ENVIRONMENT": "production",
    },
    pass_result_as_json=True,  # Pass result to stdin
    working_directory="./scripts",
)

Shell Script 예시¶

실무 운영 가이드에서 pass_result_as_json=True, True을(를) 다루는 항목입니다:

#!/bin/bash
# scripts/process_result.sh

# Read JSON from stdin
result=$(cat)

# Parse with jq and exported environment variables
status=$(echo $result | jq -r '.status')
checkpoint=$(echo $result | jq -r '.checkpoint_name')
issues="${TRUTHOUND_TOTAL_ISSUES:-0}"

echo "Checkpoint: $checkpoint"
echo "Status: $status"
echo "Issues: $issues"

# Conditional processing
if [ "$status" = "failure" ]; then
    curl -X POST "https://api.example.com/alert" \
        -H "Content-Type: application/json" \
        -d "{\"checkpoint\": \"$checkpoint\", \"issues\": $issues}"
fi

실무 운영 가이드에서 Python을(를) 다루는 항목입니다:

#!/usr/bin/env python3
# scripts/process_result.py

import json
import sys

# Read result from stdin
result = json.load(sys.stdin)

checkpoint = result["checkpoint_name"]
status = result["status"]
issue_count = len(result["validation_run"]["issues"]) if result.get("validation_run") else 0
pass_rate = result.get("validation_view", {}).get("statistics", {}).get("pass_rate")

print(f"Processing {checkpoint}: {status}")
print(f"Issues: {issue_count}")
print(f"Pass rate: {pass_rate}")

# Custom logic...

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

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

Conditional Execution¶

def conditional_callback(checkpoint_result):
    """Logic that executes only under specific conditions."""
    validation = checkpoint_result.validation_view
    stats = validation.statistics if validation else None

    # Page only when 10+ critical issues
    if stats and stats.critical_issues >= 10:
        page_on_call_engineer(checkpoint_result)
        return {"paged": True}

    return {"paged": False}


action = CustomAction(
    callback=conditional_callback,
    notify_on="failure",  # Callback invoked only on failure
)

Error Handling¶

실무 운영 가이드에서 ActionResult, ERROR을(를) 다루는 항목입니다:

def risky_callback(checkpoint_result):
    try:
        # Risky operation
        result = do_something_risky()
        return {"success": True, "result": result}
    except Exception as e:
        # Re-raising the exception records it in ActionResult.error
        raise RuntimeError(f"Failed to process: {e}")


# Or return failure explicitly
def safe_callback(checkpoint_result):
    try:
        result = do_something_risky()
        return {"success": True}
    except Exception as e:
        # Catch exception and return failure info
        return {"success": False, "error": str(e)}

Combining with Other Actions¶

from truthound.checkpoint import Checkpoint
from truthound.checkpoint.actions import (
    StoreValidationResult,
    SlackNotification,
    CustomAction,
)

def post_process(result):
    """Execute after all other actions complete."""
    # Post-process results
    aggregate_metrics(result)
    update_dashboard(result)
    return {"post_processed": True}


checkpoint = Checkpoint(
    name="my_check",
    data_source="data.csv",
    validators=["null"],
    actions=[
        # Executed in order
        StoreValidationResult(store_path="./results"),      # 1. Store
        SlackNotification(webhook_url="...", notify_on="failure"),  # 2. Notify
        CustomAction(callback=post_process),                # 3. Post-process
    ],
)

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

YAML 설정 예시¶

actions:
  # Shell command
  - type: custom
    shell_command: ./scripts/notify.sh
    environment:
      API_KEY: ${API_KEY}
    pass_result_as_json: true
    notify_on: failure

  # Python script
  - type: custom
    shell_command: python ./scripts/process.py
    working_directory: ./scripts
    pass_result_as_json: true
    notify_on: always

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