체크포인트 & CI/CD 통합 (Phase 6)¶
실무 운영 가이드에서 Truthound, Checkpoint, CI/CD을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
테이블 of Contents¶
- 개요
- 빠른 시작
- 실무 운영 가이드에서 Core, Components을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
- 실무 운영 가이드에서 Actions을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
- 실무 운영 가이드에서 Triggers을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
- 실무 운영 가이드에서 Async, Execution을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
- 실무 운영 가이드에서 Transaction, Management을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
- CI/CD 통합
- CI 리포터
- 체크포인트Runner
- 실무 운영 가이드에서 Registry을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
- Advanced 알림
- 권장 방식
- API 레퍼런스
- 실무 운영 가이드에서 Enterprise, Assessment을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
개요¶
실무 운영 가이드에서 Checkpoints을(를) 다루는 항목입니다:
- 실무 운영 가이드에서 Automated, Validation, Pipelines, Define을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
- 실무 운영 가이드에서 CI/CD, Platform, Support, Native을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
- 실무 운영 가이드에서 Async, Execution, Non-blocking을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
- 실무 운영 가이드에서 Transaction, Management, Saga을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
- 실무 운영 가이드에서 Flexible, Triggers, Schedule, Cron, Event, File-based을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
아키텍처¶
┌─────────────────────────────────────────────────────────────────────┐
│ Checkpoint │
├─────────────────────────────────────────────────────────────────────┤
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌────────────┐ │
│ │ DataSource │ │ Validators │ │ Actions │ │ Triggers │ │
│ └──────┬──────┘ └──────┬──────┘ └──────┬──────┘ └─────┬──────┘ │
│ │ │ │ │ │
│ └────────────────┼────────────────┼────────────────┘ │
│ ▼ ▼ │
│ ┌─────────────────────────────────────────────────────────────────┐│
│ │ CheckpointRunner ││
│ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────────────────┐ ││
│ │ │ Sync │ │ Async │ │ Transaction Coordinator │ ││
│ │ │ Execution │ │ Execution │ │ (Saga + Idempotency) │ ││
│ │ └─────────────┘ └─────────────┘ └─────────────────────────┘ ││
│ └─────────────────────────────────────────────────────────────────┘│
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────────────────┐│
│ │ CI/CD Reporters ││
│ │ GitHub │ GitLab │ Jenkins │ CircleCI │ Azure │ Bitbucket │ ... ││
│ └─────────────────────────────────────────────────────────────────┘│
└─────────────────────────────────────────────────────────────────────┘
실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
빠른 시작¶
Basic Usage¶
from truthound.checkpoint import Checkpoint
from truthound.checkpoint.actions import (
StoreValidationResult,
SlackNotification,
)
# Create a checkpoint
checkpoint = Checkpoint(
name="daily_user_validation",
data_source="users.csv",
validators=["null", "duplicate", "range"],
actions=[
StoreValidationResult(store_path="./results"),
SlackNotification(
webhook_url="https://hooks.slack.com/...",
notify_on="failure",
channel="#data-quality"
),
],
)
# Run the checkpoint
result = checkpoint.run()
print(result.summary())
CLI Usage¶
# Initialize a sample configuration
truthound checkpoint init -o truthound.yaml
# Run a checkpoint from config
truthound checkpoint run daily_data_validation --config truthound.yaml
# Run ad-hoc checkpoint
truthound checkpoint run quick_check --data data.csv --validators null,duplicate
# List checkpoints
truthound checkpoint list --config truthound.yaml
# Validate configuration
truthound checkpoint validate truthound.yaml
실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
Core Components¶
체크포인트Config¶
from truthound.checkpoint import Checkpoint, CheckpointConfig
config = CheckpointConfig(
name="production_validation",
data_source="s3://bucket/data.parquet",
validators=["null", "duplicate", "range"],
min_severity="medium",
schema="schema.yaml",
auto_schema=False,
run_name_template="%Y%m%d_%H%M%S",
tags={"env": "production", "team": "data-platform"},
metadata={"owner": "data-team@company.com"},
fail_on_critical=True,
fail_on_high=False,
timeout_seconds=3600,
sample_size=100000,
)
checkpoint = Checkpoint(config=config)
YAML 설정¶
# truthound.yaml
checkpoints:
- name: daily_data_validation
data_source: data/production.csv
validators:
- 'null'
- duplicate
- range
- regex
validator_config:
regex:
patterns:
email: ^[\w.+-]+@[\w-]+\.[\w.-]+$
product_code: ^[A-Z]{2,4}[-_][0-9]{3,6}$
phone: ^(\+\d{1,3}[-.\s]?)?\(?\d{3}\)?[-.\s]?\d{3}[-.\s]?\d{4}$
range:
columns:
age:
min_value: 0
max_value: 150
price:
min_value: 0
min_severity: medium
auto_schema: true
tags:
environment: production
team: data-platform
actions:
- type: store_result
store_path: ./truthound_results
partition_by: date
- type: update_docs
site_path: ./truthound_docs
include_history: true
- type: slack
webhook_url: https://hooks.slack.com/services/YOUR/WEBHOOK/URL
notify_on: failure
channel: '#data-quality'
triggers:
- type: schedule
interval_hours: 24
run_on_weekdays: [0, 1, 2, 3, 4] # Mon-Fri
실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
Actions¶
실무 운영 가이드에서 Actions, They을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
Available Actions¶
| 실무 운영 가이드에서 Action을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 Description을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 주요 기능 |
|---|---|---|
실무 운영 가이드에서 StoreValidationResult, StoreValidationResult을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
Save 결과 to filesystem, S3, or GCS | 실무 운영 가이드에서 Partitioning을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
실무 운영 가이드에서 UpdateDataDocs, UpdateDataDocs을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
실무 운영 가이드에서 HTML, Generate, HTML/Markdown을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 History을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
실무 운영 가이드에서 SlackNotification, SlackNotification을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
실무 운영 가이드에서 Send, Slack을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 Mentions을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
실무 운영 가이드에서 EmailNotification, EmailNotification을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
실무 운영 가이드에서 Send을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 SMTP, SendGrid, SES을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
실무 운영 가이드에서 WebhookAction, WebhookAction을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
실무 운영 가이드에서 Call, HTTP을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 Auth을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
실무 운영 가이드에서 PagerDutyAction, PagerDutyAction을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
실무 운영 가이드에서 Create/resolve, PagerDuty을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 Auto-resolve을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
실무 운영 가이드에서 GitHubAction, GitHubAction을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
GitHub Actions 통합 | 실무 운영 가이드에서 Summaries을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
실무 운영 가이드에서 TeamsNotification, TeamsNotification을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
실무 운영 가이드에서 Microsoft, Teams을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 Adaptive, Cards을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
실무 운영 가이드에서 OpsGenieAction, OpsGenieAction을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
OpsGenie 알림 management | 실무 운영 가이드에서 Priorities을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
실무 운영 가이드에서 DiscordNotification, DiscordNotification을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
실무 운영 가이드에서 Discord을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 Embeds을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
실무 운영 가이드에서 TelegramNotification, TelegramNotification을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
실무 운영 가이드에서 Telegram을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 Markdown을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
실무 운영 가이드에서 CustomAction, CustomAction을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
실무 운영 가이드에서 Execute, Python을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 Full을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
Store검증Result¶
from truthound.checkpoint.actions import StoreValidationResult
action = StoreValidationResult(
store_path="./results", # Local path, s3://, or gs://
store_type="file", # file, s3, gcs
format="json", # json, yaml
partition_by="date", # date, checkpoint, status
retention_days=30,
compress=True,
)
UpdateDataDocs¶
from truthound.checkpoint.actions import UpdateDataDocs
action = UpdateDataDocs(
site_path="./docs",
format="html", # html, markdown
include_history=True,
max_history_items=100,
template="default", # default, minimal, detailed
)
SlackNotification¶
from truthound.checkpoint.actions import SlackNotification
action = SlackNotification(
webhook_url="https://hooks.slack.com/...",
channel="#data-quality",
notify_on="failure", # always, success, failure, error
mention_on_failure=["U12345", "@here"],
include_details=True,
custom_message="Data quality check completed",
)
WebhookAction¶
from truthound.checkpoint.actions import WebhookAction
action = WebhookAction(
url="https://api.example.com/webhook",
method="POST",
auth_type="bearer", # none, basic, bearer, api_key
auth_credentials={"token": "${API_TOKEN}"},
headers={"X-Custom-Header": "value"},
include_full_result=True,
timeout_seconds=30,
retry_count=3,
)
TeamsNotification¶
실무 운영 가이드에서 Microsoft, Teams, Adaptive, Cards을(를) 다루는 항목입니다:
from truthound.checkpoint.actions import (
TeamsNotification,
TeamsConfig,
AdaptiveCardBuilder,
MessageTheme,
create_teams_notification,
create_failure_alert,
)
# Basic usage
action = TeamsNotification(
webhook_url="https://outlook.office.com/webhook/...",
notify_on="failure",
channel="Data Quality",
include_details=True,
)
# With custom Adaptive Card
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,
theme=MessageTheme.CRITICAL,
)
# Factory functions
action = create_failure_alert(
webhook_url="...",
mention_users=["user@company.com"],
)
OpsGenieAction¶
실무 운영 가이드에서 OpsGenie을(를) 다루는 항목입니다:
from truthound.checkpoint.actions import (
OpsGenieAction,
OpsGenieConfig,
AlertPriority,
ResponderType,
Responder,
create_opsgenie_action,
create_critical_alert,
create_team_alert,
)
# Basic usage
action = OpsGenieAction(
api_key="${OPSGENIE_API_KEY}",
notify_on="failure",
priority=AlertPriority.P1,
tags=["data-quality", "production"],
)
# With responders
action = OpsGenieAction(
api_key="...",
responders=[
Responder(type=ResponderType.TEAM, name="data-platform"),
Responder(type=ResponderType.USER, username="oncall@company.com"),
],
visible_to=[
Responder(type=ResponderType.TEAM, name="engineering"),
],
auto_resolve_on_success=True,
)
# Factory functions
action = create_critical_alert(
api_key="...",
team="data-platform",
escalation_policy="data-quality-escalation",
)
DiscordNotification¶
실무 운영 가이드에서 Discord을(를) 다루는 항목입니다:
from truthound.checkpoint.actions import DiscordNotification, DiscordConfig
action = DiscordNotification(
webhook_url="https://discord.com/api/webhooks/...",
notify_on="failure",
username="Truthound Bot",
avatar_url="https://example.com/logo.png",
embed_color=0xFF0000, # Red for errors
include_mentions=["@here"],
)
# With 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¶
실무 운영 가이드에서 Telegram을(를) 다루는 항목입니다:
from truthound.checkpoint.actions import (
TelegramNotification,
TelegramConfig,
TelegramNotificationWithPhoto,
)
# Basic text notification
action = TelegramNotification(
bot_token="${TELEGRAM_BOT_TOKEN}",
chat_id="-1001234567890", # Channel/group ID
notify_on="failure",
parse_mode="Markdown", # or "HTML"
)
# With photo (e.g., chart screenshot)
action = TelegramNotificationWithPhoto(
bot_token="...",
chat_id="...",
photo_url="https://example.com/chart.png",
caption="Data quality trend chart",
)
# Custom message template
action = TelegramNotification(
bot_token="...",
chat_id="...",
message_template="""
🚨 *Data Quality Alert*
Dataset: `{checkpoint_name}`
Status: {status}
Issues: {issue_count}
View Report: `{report_url}`
""",
)
CustomAction¶
from truthound.checkpoint.actions import CustomAction
# Python callback
def my_callback(result):
print(f"Checkpoint completed: {result.status}")
if result.status == "failure":
# Custom alerting logic
send_custom_alert(result)
return {"processed": True}
action = CustomAction(callback=my_callback)
# Shell command
action = CustomAction(
shell_command="./scripts/notify.sh",
environment={"API_KEY": "${SECRET_KEY}"},
pass_result_as_json=True,
)
Notify Conditions¶
실무 운영 가이드에서 notify_on을(를) 다루는 항목입니다:
| 실무 운영 가이드에서 Condition을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 Triggers을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
|---|---|
실무 운영 가이드에서 always을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
실무 운영 가이드에서 success을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
검증 passed |
실무 운영 가이드에서 failure을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
검증 failed |
실무 운영 가이드에서 error을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
실무 운영 가이드에서 System을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
실무 운영 가이드에서 failure_or_error을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
실패 or error |
실무 운영 가이드에서 not_success을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
실무 운영 가이드에서 Any을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
Triggers¶
실무 운영 가이드에서 Triggers을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
ScheduleTrigger¶
실무 운영 가이드에서 Time-interval을(를) 다루는 항목입니다:
from truthound.checkpoint.triggers import ScheduleTrigger
# Run every hour
trigger = ScheduleTrigger(interval_hours=1)
# Run every 30 minutes on weekdays
trigger = ScheduleTrigger(
interval_minutes=30,
run_on_weekdays=[0, 1, 2, 3, 4], # Mon=0, Sun=6
start_time=datetime(2024, 1, 1, 9, 0), # Start at 9 AM
end_time=datetime(2024, 12, 31, 18, 0), # End at 6 PM
timezone="America/New_York",
)
CronTrigger¶
실무 운영 가이드에서 Standard을(를) 다루는 항목입니다:
from truthound.checkpoint.triggers import CronTrigger
# Daily at midnight
trigger = CronTrigger(expression="0 0 * * *")
# Every 15 minutes
trigger = CronTrigger(expression="*/15 * * * *")
# Monday at 9am
trigger = CronTrigger(expression="0 9 * * 1")
# With seconds (6 fields)
trigger = CronTrigger(expression="30 0 9 * * 1") # Monday 9:00:30
EventTrigger¶
실무 운영 가이드에서 Event-driven을(를) 다루는 항목입니다:
from truthound.checkpoint.triggers import EventTrigger
trigger = EventTrigger(
event_type="data_updated",
event_filter={"source": "production", "priority": "high"},
debounce_seconds=60, # Minimum time between triggers
batch_events=True, # Batch multiple events
batch_window_seconds=30, # Batch window
)
# Fire event programmatically
trigger.fire_event({
"source": "production",
"priority": "high",
"table": "users",
"rows_affected": 1500,
})
FileWatchTrigger¶
실무 운영 가이드에서 File을(를) 다루는 항목입니다:
from truthound.checkpoint.triggers import FileWatchTrigger
trigger = FileWatchTrigger(
paths=["./data", "/shared/datasets"],
patterns=["*.csv", "*.parquet"],
recursive=True,
events=["modified", "created"], # modified, created, deleted
ignore_patterns=[".*", "__pycache__", "*.tmp"],
hash_check=True, # Only trigger on content change
poll_interval_seconds=5,
)
실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
Async Execution¶
실무 운영 가이드에서 AsyncCheckpoint, AsyncCheckpoint을(를) 다루는 항목입니다:
import asyncio
from truthound.checkpoint import AsyncCheckpoint
from truthound.checkpoint.async_actions import AsyncSlackNotification
# Create async checkpoint
checkpoint = AsyncCheckpoint(
name="async_validation",
data_source="large_dataset.parquet",
validators=["null", "duplicate"],
actions=[
AsyncSlackNotification(webhook_url="..."),
],
max_concurrent_actions=5,
execution_strategy="concurrent", # sequential, concurrent, pipeline
)
# Run asynchronously
async def main():
result = await checkpoint.run_async()
print(result.summary())
asyncio.run(main())
Execution Strategies¶
from truthound.checkpoint.async_base import (
SequentialStrategy,
ConcurrentStrategy,
PipelineStrategy,
)
# Sequential: One action at a time
checkpoint = AsyncCheckpoint(
execution_strategy=SequentialStrategy()
)
# Concurrent: All actions in parallel with limit
checkpoint = AsyncCheckpoint(
execution_strategy=ConcurrentStrategy(max_concurrency=10)
)
# Pipeline: Staged execution
# Stage 1: Store result and update docs (parallel)
# Stage 2: Notify (after stage 1)
checkpoint = AsyncCheckpoint(
execution_strategy=PipelineStrategy(
stages=[[0, 1], [2]] # Action indices
)
)
Running Multiple 체크포인트 Concurrently¶
from truthound.checkpoint import run_checkpoints_async
checkpoints = [checkpoint1, checkpoint2, checkpoint3]
results = await run_checkpoints_async(
checkpoints,
max_concurrent=5,
context={"batch_id": "2024-01-15"},
)
for result in results:
print(f"{result.checkpoint_name}: {result.status}")
Async Callbacks¶
async def on_complete(result):
await send_metrics_async(result.to_dict())
async def on_error(result):
await alert_team_async(result.error)
checkpoint = AsyncCheckpoint(
name="monitored_check",
on_complete=on_complete,
on_error=on_error,
)
실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
Transaction Management¶
실무 운영 가이드에서 Truthound, Saga을(를) 다루는 항목입니다:
Compensatable Actions¶
실무 운영 가이드에서 Actions을(를) 다루는 항목입니다:
from truthound.checkpoint.transaction import CompensatableAction
class DatabaseUpdateAction(CompensatableAction):
def execute(self, result):
# Forward action
self.backup_id = create_backup()
update_database(result)
return ActionResult(status="success")
def compensate(self, result, execute_result):
# Rollback action
restore_from_backup(self.backup_id)
return ActionResult(status="compensated")
Transaction Coordinator¶
from truthound.checkpoint.transaction import TransactionCoordinator
coordinator = TransactionCoordinator(
actions=[action1, action2, action3],
compensation_strategy="reverse", # reverse, parallel
max_compensation_retries=3,
)
result = coordinator.execute(checkpoint_result)
if result.needs_rollback:
coordinator.compensate(result)
Idempotency¶
실무 운영 가이드에서 Prevent을(를) 다루는 항목입니다:
from truthound.checkpoint.idempotency import IdempotencyService
service = IdempotencyService(
store="redis://localhost:6379", # Or filesystem, memory
ttl_seconds=3600,
)
# Check before execution
idempotency_key = f"checkpoint:{name}:{run_id}"
if service.is_duplicate(idempotency_key):
return cached_result
result = checkpoint.run()
service.mark_completed(idempotency_key, result)
실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
CI/CD 통합¶
GitHub Actions¶
# .github/workflows/data-quality.yml
name: Data Quality Check
on:
schedule:
- cron: '0 0 * * *'
push:
paths:
- 'data/**'
pull_request:
paths:
- 'data/**'
jobs:
validate:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- uses: actions/setup-python@v5
with:
python-version: '3.11'
- name: Install Truthound
run: pip install truthound[all]
- name: Run Validation
run: |
truthound checkpoint run daily_data_validation \
--config truthound.yaml \
--github-summary \
--strict
env:
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
SLACK_WEBHOOK: ${{ secrets.SLACK_WEBHOOK }}
- name: Upload Results
if: always()
uses: actions/upload-artifact@v4
with:
name: data-quality-report
path: truthound_results/
GitLab CI¶
# .gitlab-ci.yml
stages:
- validate
data-quality:
stage: validate
image: python:3.11-slim
variables:
PIP_CACHE_DIR: "$CI_PROJECT_DIR/.pip-cache"
cache:
paths:
- .pip-cache/
script:
- pip install truthound[all]
- truthound checkpoint run $CHECKPOINT_NAME --config truthound.yaml
artifacts:
when: always
paths:
- truthound_results/
- truthound_docs/
reports:
dotenv: truthound.env
rules:
- if: $CI_PIPELINE_SOURCE == "schedule"
- if: $CI_PIPELINE_SOURCE == "merge_request_event"
changes:
- data/**/*
Jenkins¶
// Jenkinsfile
pipeline {
agent any
environment {
SLACK_WEBHOOK = credentials('slack-webhook')
}
stages {
stage('Setup') {
steps {
sh 'pip install truthound[all]'
}
}
stage('Data Quality') {
steps {
sh '''
truthound checkpoint run daily_data_validation \
--config truthound.yaml \
--format json \
--output truthound-result.json
'''
}
post {
always {
archiveArtifacts artifacts: 'truthound-result.json'
junit 'truthound-junit.xml'
}
failure {
slackSend channel: '#data-quality',
message: "Data Quality Check Failed: ${env.BUILD_URL}"
}
}
}
}
}
CircleCI¶
# .circleci/config.yml
version: 2.1
jobs:
data-quality:
docker:
- image: cimg/python:3.11
steps:
- checkout
- run:
name: Install Dependencies
command: pip install truthound[all]
- run:
name: Run Validation
command: |
truthound checkpoint run daily_data_validation \
--config truthound.yaml \
--format json
- store_test_results:
path: test-results/truthound
- store_artifacts:
path: artifacts
workflows:
nightly:
triggers:
- schedule:
cron: "0 0 * * *"
filters:
branches:
only: main
jobs:
- data-quality
Azure DevOps¶
# azure-pipelines.yml
trigger:
paths:
include:
- data/*
schedules:
- cron: "0 0 * * *"
displayName: Daily midnight run
branches:
include:
- main
pool:
vmImage: 'ubuntu-latest'
steps:
- task: UsePythonVersion@0
inputs:
versionSpec: '3.11'
- script: pip install truthound[all]
displayName: 'Install Truthound'
- script: |
truthound checkpoint run $(CHECKPOINT_NAME) \
--config truthound.yaml \
--format json
displayName: 'Run Data Quality Check'
env:
SLACK_WEBHOOK: $(SLACK_WEBHOOK)
- publish: truthound_results
artifact: DataQualityReport
condition: always()
Generate CI Configs¶
from truthound.checkpoint.ci import (
generate_github_workflow,
generate_gitlab_ci,
generate_jenkinsfile,
generate_circleci_config,
write_ci_config,
)
# Generate GitHub Actions workflow
workflow = generate_github_workflow(
checkpoint_name="daily_data_validation",
schedule="0 0 * * *",
notify_slack=True,
python_version="3.11",
)
# Generate all configs
write_ci_config("github", checkpoint_name="daily_data_validation")
write_ci_config("gitlab", checkpoint_name="daily_data_validation")
write_ci_config("jenkins", checkpoint_name="daily_data_validation")
실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
CI 리포터¶
실무 운영 가이드에서 Truthound을(를) 다루는 항목입니다:
Supported 플랫폼¶
| 플랫폼 | 실무 운영 가이드에서 Detection을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 Features을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
|---|---|---|
| 실무 운영 가이드에서 GitHub, Actions을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 GITHUB_ACTIONS=true, GITHUB_ACTIONS을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
실무 운영 가이드에서 Step, Summary, Annotations, Outputs을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
| 실무 운영 가이드에서 GitLab을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 GITLAB_CI=true, GITLAB_CI을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
dotenv 아티팩트, ANSI colors |
| 실무 운영 가이드에서 Jenkins을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 JENKINS_URL, JENKINS_URL을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
JUnit XML, Properties 파일 |
| 실무 운영 가이드에서 CircleCI을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 CIRCLECI=true, CIRCLECI을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
test-결과, 아티팩트 |
| 실무 운영 가이드에서 Azure, DevOps을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 TF_BUILD=True, TF_BUILD, True을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
실무 운영 가이드에서 Build, Variables을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
| Bitbucket 파이프라인 | 실무 운영 가이드에서 BITBUCKET_BUILD_NUMBER, BITBUCKET_BUILD_NUMBER을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
실무 운영 가이드에서 Pipes을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
| 실무 운영 가이드에서 Travis을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 TRAVIS=true, TRAVIS을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
실무 운영 가이드에서 Environment을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
| 실무 운영 가이드에서 TeamCity을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 TEAMCITY_VERSION, TEAMCITY_VERSION을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
실무 운영 가이드에서 Service을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
| 실무 운영 가이드에서 Buildkite을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 BUILDKITE=true, BUILDKITE을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
실무 운영 가이드에서 API, Annotations을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
| 실무 운영 가이드에서 Drone을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 DRONE=true, DRONE을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
실무 운영 가이드에서 Environment을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
| 실무 운영 가이드에서 AWS, CodeBuild을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 CODEBUILD_BUILD_ID, CODEBUILD_BUILD_ID을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
실무 운영 가이드에서 BuildSpec을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
| 실무 운영 가이드에서 GCP, Cloud, Build을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 BUILDER_OUTPUT, BUILDER_OUTPUT을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
실무 운영 가이드에서 Environment을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
Using CI 리포터¶
from truthound.checkpoint.ci import (
detect_ci_platform,
get_ci_environment,
get_ci_reporter,
is_ci_environment,
)
# Check if in CI
if is_ci_environment():
env = get_ci_environment()
print(f"Platform: {env.platform}")
print(f"Repository: {env.repository}")
print(f"Branch: {env.branch}")
print(f"Commit: {env.commit_sha}")
print(f"PR Number: {env.pr_number}")
print(f"Run URL: {env.run_url}")
# Get platform-specific reporter
reporter = get_ci_reporter()
reporter.report_status(result)
reporter.set_output("total_issues", stats.total_issues)
reporter.set_output("status", result.status.value)
GitHub Actions Reporter¶
from truthound.reporters.ci import GitHubActionsReporter
reporter = GitHubActionsReporter(
step_summary=True, # Write to GITHUB_STEP_SUMMARY
use_groups=True, # Use ::group:: for collapsible sections
emoji_enabled=True, # Include emojis in output
set_output=True, # Set workflow outputs
)
# Report to GitHub
exit_code = reporter.report_to_ci(result)
# The reporter automatically:
# - Writes job summary in Markdown
# - Emits annotations (::error::, ::warning::, ::notice::)
# - Sets output variables via GITHUB_OUTPUT
Custom Annotations¶
from truthound.reporters.ci.base import CIAnnotation, AnnotationLevel
annotation = CIAnnotation(
message="Null values exceed threshold (15% > 5%)",
level=AnnotationLevel.ERROR,
file="data/users.csv",
line=42,
title="Data Quality Issue",
validator_name="NullValidator",
)
reporter.format_annotation(annotation)
# Output: ::error file=data/users.csv,line=42,title=Data Quality Issue::Null values exceed threshold (15% > 5%)
실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
체크포인트Runner¶
실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 다루는 항목입니다:
from truthound.checkpoint import Checkpoint, CheckpointRunner
from truthound.checkpoint.triggers import ScheduleTrigger, CronTrigger
# Create checkpoints with triggers
hourly_metrics_check = Checkpoint(
name="hourly_metrics_check",
data_source="data.csv",
validators=["null", "duplicate"],
).add_trigger(ScheduleTrigger(interval_hours=1))
daily_data_validation = Checkpoint(
name="daily_data_validation",
data_source="data.parquet",
validators=["range", "distribution"],
).add_trigger(CronTrigger(expression="0 0 * * *"))
# Create runner
runner = CheckpointRunner(
max_workers=4,
result_callback=lambda r: print(f"Completed: {r.checkpoint_name}"),
error_callback=lambda e: print(f"Error: {e}"),
)
# Add checkpoints
runner.add_checkpoint(hourly_metrics_check)
runner.add_checkpoint(daily_data_validation)
# Start background execution
runner.start()
# Run specific checkpoint once
result = runner.run_once("hourly_metrics_check")
# Run all checkpoints
results = runner.run_all()
# Iterate over results
for result in runner.iter_results(timeout=60):
print(result.summary())
# Stop runner
runner.stop()
실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
Registry¶
실무 운영 가이드에서 Register을(를) 다루는 항목입니다:
from truthound.checkpoint import (
Checkpoint,
CheckpointRegistry,
register_checkpoint,
get_checkpoint,
list_checkpoints,
load_checkpoints,
)
# Create registry
registry = CheckpointRegistry()
# Register checkpoints
checkpoint = Checkpoint(name="my_check", data_source="data.csv")
registry.register(checkpoint)
# Or use global registry
register_checkpoint(checkpoint)
# Retrieve
cp = get_checkpoint("my_check")
result = cp.run()
# List all
names = list_checkpoints()
print(names) # ['my_check', ...]
# Load from file
checkpoints = load_checkpoints("truthound.yaml")
for cp in checkpoints:
registry.register(cp)
# Check existence
if "my_check" in registry:
print("Checkpoint exists")
실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
Advanced 알림¶
실무 운영 가이드에서 Truthound을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
Rule-based Routing¶
실무 운영 가이드에서 Route을(를) 다루는 항목입니다:
from truthound.checkpoint.routing import ActionRouter, SeverityRule, Route
from truthound.checkpoint.actions import SlackNotification, PagerDutyAction
router = ActionRouter()
# Critical alerts go to PagerDuty
router.add_route(Route(
name="critical",
rule=SeverityRule(min_severity="critical"),
actions=[PagerDutyAction(service_key="...")],
priority=1,
))
# High severity goes to Slack
router.add_route(Route(
name="high",
rule=SeverityRule(min_severity="high"),
actions=[SlackNotification(webhook_url="...")],
priority=2,
))
# Use with checkpoint
checkpoint = Checkpoint(
name="daily_data_validation",
data_source="data.csv",
router=router,
)
실무 운영 가이드에서 Available, Rules, SeverityRule, IssueCountRule, StatusRule, TagRule, PassRateRule, TimeWindowRule을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
실무 운영 가이드에서 AllOf, AnyOf, NotRule, Combinators, AllOf, AnyOf, NotRule을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
Notification Deduplication¶
실무 운영 가이드에서 Prevent을(를) 다루는 항목입니다:
from truthound.checkpoint.deduplication import (
NotificationDeduplicator,
InMemoryDeduplicationStore,
TimeWindow,
)
deduplicator = NotificationDeduplicator(
store=InMemoryDeduplicationStore(),
default_window=TimeWindow(seconds=300), # 5 minutes
)
fingerprint = deduplicator.generate_fingerprint(
checkpoint_name="daily_data_validation",
action_type="slack",
severity="high",
)
if not deduplicator.is_duplicate(fingerprint):
await action.execute(result)
deduplicator.mark_sent(fingerprint)
실무 운영 가이드에서 Window, Strategies, Sliding, Tumbling, Session, Adaptive을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
실무 운영 가이드에서 Storage, Backends, InMemory, Redis, Streams을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
Rate Limiting / Throttling¶
실무 운영 가이드에서 Control을(를) 다루는 항목입니다:
from truthound.checkpoint.throttling import ThrottlerBuilder, ThrottlingMiddleware
throttler = (
ThrottlerBuilder()
.with_per_minute_limit(10)
.with_per_hour_limit(100)
.with_per_day_limit(500)
.build()
)
middleware = ThrottlingMiddleware(throttler=throttler)
throttled_action = middleware.wrap(slack_action)
실무 운영 가이드에서 Algorithms, Token, Bucket, Fixed, Window, Sliding, Composite을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
Escalation Policies¶
Multi-level 알림 escalation:
from truthound.checkpoint.escalation import (
EscalationPolicy,
EscalationLevel,
EscalationEngine,
)
policy = EscalationPolicy(
name="critical_alerts",
levels=[
EscalationLevel(level=1, delay_minutes=0, targets=["team-lead"]),
EscalationLevel(level=2, delay_minutes=15, targets=["manager"]),
EscalationLevel(level=3, delay_minutes=30, targets=["director"]),
],
)
engine = EscalationEngine(policy=policy)
await engine.trigger("incident-123", context={"severity": "critical"})
# Later: acknowledge or resolve
await engine.acknowledge("incident-123", acknowledged_by="john@company.com")
await engine.resolve("incident-123", resolved_by="jane@company.com")
실무 운영 가이드에서 States, PENDING, TRIGGERED, ACKNOWLEDGED, ESCALATED, RESOLVED을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
실무 운영 가이드에서 SQLite, SQL, Storage, Backends, InMemory, Redis을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
권장 방식¶
1. Use 설정 파일¶
실무 운영 가이드에서 YAML, Store을(를) 다루는 항목입니다:
# truthound.yaml
checkpoints:
- name: production_daily
data_source: ${DATA_PATH} # Use environment variables
validators:
- "null"
- duplicate
actions:
- type: store_result
store_path: ${RESULTS_PATH}
2. Set Up Appropriate 알림¶
actions = [
# Always store results for audit
StoreValidationResult(notify_on="always"),
# Update docs on success
UpdateDataDocs(notify_on="success"),
# Alert only on failures
SlackNotification(notify_on="failure"),
PagerDutyAction(notify_on="failure_or_error"),
]
3. Use Strict Mode in CI¶
실무 운영 가이드에서 --strict을(를) 다루는 항목입니다:
- 실무 운영 가이드에서 Any을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
- The 체크포인트 status is "실패" or "error"
실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
실무 운영 가이드 개요¶
# For large datasets, use async execution
checkpoint = AsyncCheckpoint(
name="large_data_check",
data_source="large_dataset.parquet",
sample_size=100000, # Sample for faster validation
max_concurrent_actions=10,
)
result = await checkpoint.run_async()
5. Implement Idempotency for Production¶
from truthound.checkpoint.idempotency import IdempotencyService
service = IdempotencyService(store="redis://localhost:6379")
# Prevent duplicate runs
if not service.is_duplicate(run_key):
result = checkpoint.run()
service.mark_completed(run_key, result)
6. Monitor Trends Over Time¶
StoreValidationResult(
store_path="s3://bucket/dq-results",
partition_by="date",
retention_days=90,
)
실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
API 레퍼런스¶
체크포인트Result¶
result = checkpoint.run()
result.run_id # Unique run identifier
result.checkpoint_name # Checkpoint name
result.run_time # When the checkpoint ran
result.status # CheckpointStatus (success/failure/error/warning)
result.validation_run # ValidationRunResult from check()
result.validation_view # Compatibility statistics/results projection
result.action_results # List of ActionResult
result.duration_ms # Execution duration in milliseconds
result.error # Error message if failed
result.metadata # Custom metadata dict
# Methods
result.to_dict() # Serialize to dictionary
result.from_dict(d) # Deserialize from dictionary
result.summary() # Human-readable summary string
체크포인트Status¶
from truthound.checkpoint.checkpoint import CheckpointStatus
CheckpointStatus.SUCCESS # All validations passed
CheckpointStatus.FAILURE # Validation failures detected
CheckpointStatus.WARNING # Non-critical issues found
CheckpointStatus.ERROR # System error occurred
ActionResult¶
from truthound.checkpoint.actions.base import ActionResult, ActionStatus
result = ActionResult(
action_name="slack_notification",
action_type="notification",
status=ActionStatus.SUCCESS,
message="Notification sent successfully",
started_at=datetime.now(),
completed_at=datetime.now(),
duration_ms=150.5,
details={"message_id": "abc123"},
)
CIEnvironment¶
from truthound.checkpoint.ci import get_ci_environment
env = get_ci_environment()
env.platform # CIPlatform enum
env.is_ci # bool
env.is_pr # bool (is pull request)
env.branch # str
env.commit_sha # str
env.commit_message # str
env.pr_number # int | None
env.pr_target_branch # str
env.repository # str (owner/repo)
env.run_id # str
env.run_url # str
env.actor # str (user who triggered)
env.job_name # str
env.workflow_name # str
실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
Enterprise Assessment¶
Feature Completeness¶
| 실무 운영 가이드에서 Feature을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 Status을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 Notes을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
|---|---|---|
| Core 체크포인트 | 실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 Full을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
| 실무 운영 가이드에서 Multiple, Actions을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
| 실무 운영 가이드에서 Trigger, Types을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 스케줄, Cron, Event, FileWatch |
| 실무 운영 가이드에서 Async, Execution을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 Native을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
| 실무 운영 가이드에서 Transaction, Management을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 Saga을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
| 실무 운영 가이드에서 Idempotency을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 Duplicate을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
| 12 CI 플랫폼 | 실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 Industry-leading을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
| 실무 운영 가이드에서 JUnit, XML, Output을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 Jenkins/CI을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
| 실무 운영 가이드에서 Rule-based, Routing을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 11 rules, combinators, Python/Jinja2 엔진 |
| 실무 운영 가이드에서 Notification, Deduplication을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 InMemory/Redis을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
| 실무 운영 가이드에서 Rate, Limiting을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 Token, Bucket을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
| 실무 운영 가이드에서 Escalation, Policies을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 State을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
Code 메트릭¶
| 실무 운영 가이드에서 Metric을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 Value을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
|---|---|
| 실무 운영 가이드에서 Total, LOC을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
| 실무 운영 가이드에서 Test, LOC을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
| Test 파일 | 실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
| CI 플랫폼 | 실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
| 실무 운영 가이드에서 Action, Types을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
| 실무 운영 가이드에서 Trigger, Types을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
Comparison with Great Expectations¶
| 실무 운영 가이드에서 Feature을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 Great Expectations, Great, Expectations을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 Truthound을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
|---|---|---|
| 체크포인트 Definition | 실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
| 실무 운영 가이드에서 Multiple, Actions을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
| 스케줄 Triggers | 실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 Cron을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
| 실무 운영 가이드에서 Event, Triggers을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 Limited을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 Full을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
| 파일 Watch Triggers | 실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
| 실무 운영 가이드에서 Async, Execution을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 Native을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
| 실무 운영 가이드에서 Transaction/Saga을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
| 실무 운영 가이드에서 Idempotency을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
| CI 플랫폼 | 실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
| 실무 운영 가이드에서 JUnit, Output을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 Plugin을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 Built-in을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
| 실무 운영 가이드에서 Rule-based, Routing을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
| 실무 운영 가이드에서 Deduplication을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 InMemory/Redis을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
| 실무 운영 가이드에서 Rate, Limiting을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 Token, Bucket을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
| 실무 운영 가이드에서 Escalation, Policies을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. | 실무 운영 가이드에서 APScheduler을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다. |
실무 운영 가이드에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
함께 보기¶
- 실무 운영 가이드에서 Data, Sources, Connecting을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
- 실무 운영 가이드에서 Validators, Guide, Validator을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
- 실무 운영 가이드에서 Storage, Backends, Storing을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
- 실무 운영 가이드에서 Reporter, SDK, Output을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
- 실무 운영 가이드에서 Examples, Complete을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.