Custom Actions

Actions for executing user-defined logic.

CustomAction

Executes Python callback functions or shell commands.

Configuration

Property	Type	Default	Description
callback	Callable \| None	None	Python callback function
shell_command	str \| None	None	Shell command
environment	dict[str, str]	{}	Environment variables
pass_result_as_json	bool	True	Pass result as JSON (for shell commands)
working_directory	str \| None	None	Working directory
notify_on	str	"always"	Execution condition

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 Examples

When pass_result_as_json=True, the result is passed to stdin:

#!/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 script example:

#!/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...

When pass_result_as_json=True, Truthound also exports convenience environment variables for shell commands:

• TRUTHOUND_STATUS
• TRUTHOUND_RUN_ID
• TRUTHOUND_CHECKPOINT
• TRUTHOUND_DATA_ASSET
• TRUTHOUND_TOTAL_ISSUES
• TRUTHOUND_CRITICAL_ISSUES
• TRUTHOUND_HIGH_ISSUES
• 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

When an exception occurs in the callback, the ActionResult status becomes 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 Configuration Examples

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

Note: Python callbacks cannot be specified directly in YAML. Use the Python API for complex logic.