Notification Actions¶
Actions for sending notifications. Supports Slack, Email, Teams, Discord, and Telegram.
SlackNotification¶
Sends notifications via Slack Incoming Webhook.
Configuration (SlackConfig)¶
| Property | Type | Default | Description |
|---|---|---|---|
webhook_url |
str |
"" |
Slack Incoming Webhook URL |
channel |
str \| None |
None |
Channel override |
username |
str |
"Truthound" |
Bot display name |
icon_emoji |
str |
":mag:" |
Bot icon emoji |
include_details |
bool |
True |
Include detailed statistics |
mention_on_failure |
list[str] |
[] |
User IDs to mention on failure |
custom_message |
str \| None |
None |
Custom message template |
notify_on |
str |
"failure" |
Execution condition |
Usage Examples¶
from truthound.checkpoint.actions import SlackNotification
# Basic usage
action = SlackNotification(
webhook_url="https://hooks.slack.com/services/T00/B00/XXX",
notify_on="failure",
)
# Channel and mention configuration
action = SlackNotification(
webhook_url="https://hooks.slack.com/services/T00/B00/XXX",
channel="#data-quality",
mention_on_failure=["U12345678", "@here"], # User ID or @here/@channel
include_details=True,
notify_on="failure_or_error",
)
# Custom message
action = SlackNotification(
webhook_url="...",
custom_message="Checkpoint '{checkpoint}' {status}: {total_issues} issues found",
)
Message Format¶
Messages are sent in Block Kit format:
- Status emoji (
:white_check_mark:,:x:,:warning:) - Checkpoint name and status
- Data asset, run ID, issue count, pass rate
- Issue count by severity
- Execution time
EmailNotification¶
Sends email notifications. Supports SMTP, SendGrid, and AWS SES.
Configuration (EmailConfig)¶
| Property | Type | Default | Description |
|---|---|---|---|
smtp_host |
str |
"localhost" |
SMTP server host |
smtp_port |
int |
587 |
SMTP server port |
smtp_user |
str \| None |
None |
SMTP authentication user |
smtp_password |
str \| None |
None |
SMTP authentication password |
use_tls |
bool |
True |
Use TLS |
use_ssl |
bool |
False |
Use SSL |
from_address |
str |
"" |
Sender address |
to_addresses |
list[str] |
[] |
Recipient addresses |
cc_addresses |
list[str] |
[] |
CC addresses |
subject_template |
str |
"[Truthound] {status} - {checkpoint}" |
Subject template |
include_html |
bool |
True |
Include HTML body |
provider |
str |
"smtp" |
Provider: smtp, sendgrid, ses |
api_key |
str \| None |
None |
API key (for SendGrid, etc.) |
notify_on |
str |
"failure" |
Execution condition |
SMTP Usage¶
from truthound.checkpoint.actions import EmailNotification
action = EmailNotification(
smtp_host="smtp.gmail.com",
smtp_port=587,
smtp_user="alerts@example.com",
smtp_password="${SMTP_PASSWORD}", # Environment variable reference
use_tls=True,
from_address="alerts@example.com",
to_addresses=["team@example.com", "lead@example.com"],
cc_addresses=["manager@example.com"],
notify_on="failure",
)
SendGrid Usage¶
action = EmailNotification(
provider="sendgrid",
api_key="${SENDGRID_API_KEY}",
from_address="alerts@example.com",
to_addresses=["team@example.com"],
notify_on="failure",
)
Requirements: None (uses standard library)
AWS SES Usage¶
action = EmailNotification(
provider="ses",
from_address="alerts@example.com", # Email verified in SES
to_addresses=["team@example.com"],
notify_on="failure",
)
# AWS credentials via environment variables or IAM role
Requirements: pip install boto3
TeamsNotification¶
Sends Adaptive Card format notifications via Microsoft Teams Incoming Webhook.
Configuration¶
| Property | Type | Default | Description |
|---|---|---|---|
webhook_url |
str |
"" |
Teams Webhook URL |
channel |
str \| None |
None |
Channel name (for display) |
include_details |
bool |
True |
Include detailed information |
theme |
MessageTheme |
AUTO |
Message theme |
card_builder |
AdaptiveCardBuilder \| None |
None |
Custom card builder |
notify_on |
str |
"failure" |
Execution condition |
Usage Examples¶
from truthound.checkpoint.actions import TeamsNotification
# Basic usage
action = TeamsNotification(
webhook_url="https://outlook.office.com/webhook/...",
notify_on="failure",
)
# Theme and detail configuration
from truthound.checkpoint.actions.teams_notify import MessageTheme
action = TeamsNotification(
webhook_url="...",
channel="Data Quality",
include_details=True,
theme=MessageTheme.CRITICAL,
notify_on="failure_or_error",
)
Adaptive Card Customization¶
from truthound.checkpoint.actions.teams_notify import AdaptiveCardBuilder
builder = AdaptiveCardBuilder()
builder.add_header("Data Quality Alert")
builder.add_fact("Dataset", "users.csv")
builder.add_fact("Issues", "150")
builder.add_action_button("View Report", "https://...")
action = TeamsNotification(
webhook_url="...",
card_builder=builder,
)
DiscordNotification¶
Sends Embed format notifications via Discord Webhook.
Configuration¶
| Property | Type | Default | Description |
|---|---|---|---|
webhook_url |
str |
"" |
Discord Webhook URL |
username |
str |
"Truthound Bot" |
Bot display name |
avatar_url |
str \| None |
None |
Bot avatar URL |
embed_color |
int |
Auto | Embed color (hex integer) |
embed_title |
str \| None |
None |
Embed title |
embed_description |
str \| None |
None |
Embed description |
embed_fields |
list[dict] |
[] |
Custom fields |
include_mentions |
list[str] |
[] |
Mention list (@here, role IDs, etc.) |
notify_on |
str |
"failure" |
Execution condition |
Usage Examples¶
from truthound.checkpoint.actions import DiscordNotification
# Basic usage
action = DiscordNotification(
webhook_url="https://discord.com/api/webhooks/...",
notify_on="failure",
)
# Custom configuration
action = DiscordNotification(
webhook_url="...",
username="Data Quality Bot",
avatar_url="https://example.com/logo.png",
embed_color=0xFF0000, # Red
include_mentions=["@here"],
notify_on="failure_or_error",
)
# Custom Embed
action = DiscordNotification(
webhook_url="...",
embed_title="Data Quality Alert",
embed_description="Validation failed for users.csv",
embed_fields=[
{"name": "Issues", "value": "150", "inline": True},
{"name": "Severity", "value": "High", "inline": True},
],
)
TelegramNotification¶
Sends notifications via Telegram Bot API.
Configuration¶
| Property | Type | Default | Description |
|---|---|---|---|
bot_token |
str |
"" |
Telegram Bot Token |
chat_id |
str |
"" |
Channel/Group ID |
parse_mode |
str |
"Markdown" |
Parse mode: Markdown, HTML |
message_template |
str \| None |
None |
Custom message template |
disable_notification |
bool |
False |
Silent notification |
notify_on |
str |
"failure" |
Execution condition |
Usage Examples¶
from truthound.checkpoint.actions import TelegramNotification
# Basic usage
action = TelegramNotification(
bot_token="${TELEGRAM_BOT_TOKEN}",
chat_id="-1001234567890", # Channel/Group ID
notify_on="failure",
)
# Custom message
action = TelegramNotification(
bot_token="...",
chat_id="...",
parse_mode="Markdown",
message_template="""
*Data Quality Alert*
Dataset: `{checkpoint_name}`
Status: {status}
Issues: {issue_count}
[View Report]({report_url})
""",
)
Photo Attachment¶
from truthound.checkpoint.actions.telegram_notify import TelegramNotificationWithPhoto
action = TelegramNotificationWithPhoto(
bot_token="...",
chat_id="...",
photo_url="https://example.com/chart.png",
caption="Data quality trend chart",
)
YAML Configuration Examples¶
actions:
# Slack notification
- type: slack
webhook_url: ${SLACK_WEBHOOK_URL}
channel: "#data-quality"
mention_on_failure:
- "U12345678"
notify_on: failure
# Email notification
- type: email
smtp_host: smtp.gmail.com
smtp_port: 587
smtp_user: ${SMTP_USER}
smtp_password: ${SMTP_PASSWORD}
use_tls: true
from_address: alerts@example.com
to_addresses:
- team@example.com
- lead@example.com
notify_on: failure
# Teams notification
- type: teams
webhook_url: ${TEAMS_WEBHOOK_URL}
include_details: true
notify_on: failure_or_error
# Discord notification
- type: discord
webhook_url: ${DISCORD_WEBHOOK_URL}
include_mentions:
- "@here"
notify_on: failure
# Telegram notification
- type: telegram
bot_token: ${TELEGRAM_BOT_TOKEN}
chat_id: "-1001234567890"
notify_on: failure