Would you like me to modify this script to log the results to a CSV file or send an email
alert if a test fails?
This updated script includes a logging system that saves every test run to a CSV file and a
notification module to send an email alert if any validation step fails.
Updated Python Validation Script
import pyodbc
import csv
import smtplib
from datetime import datetime
from [Link] import EmailMessage
# --- CONFIGURATION ---
DB_CONFIG = {
"driver": "{SQL Server}",
"server": "YOUR_SERVER_NAME",
"database": "YOUR_DATABASE_NAME",
"trusted": "yes"
}
EMAIL_CONFIG = {
"sender": "alerts@[Link]",
"recipient": "data_team@[Link]",
"smtp_server": "[Link]",
"port": 587,
"password": "YOUR_APP_PASSWORD" # Use environment variables for security
}
LOG_FILE = "etl_validation_log.csv"
def send_alert(subject, body):
msg = EmailMessage()
msg.set_content(body)
msg['Subject'] = subject
msg['From'] = EMAIL_CONFIG["sender"]
msg['To'] = EMAIL_CONFIG["recipient"]
try:
with [Link](EMAIL_CONFIG["smtp_server"], EMAIL_CONFIG["port"]) as server:
[Link]()
[Link](EMAIL_CONFIG["sender"], EMAIL_CONFIG["password"])
server.send_message(msg)
print("📧 Alert email sent successfully.")
except Exception as e:
print(f"❌ Failed to send email: {e}")
def log_result(test_name, status, details):
file_exists = False
try:
with open(LOG_FILE, 'r') as f: file_exists = True
except FileNotFoundError: pass
with open(LOG_FILE, 'a', newline='') as f:
writer = [Link](f)
if not file_exists:
[Link](["Timestamp", "Test Name", "Status", "Details"])
[Link]([[Link](), test_name, status, details])
def run_etl_checks():
conn_str =
f"Driver={DB_CONFIG['driver']};Server={DB_CONFIG['server']};Database={DB_CONFIG['datab
ase']};Trusted_Connection={DB_CONFIG['trusted']};"
overall_failure = False
report_summary = []
try:
conn = [Link](conn_str)
cursor = [Link]()
# 1. Count Reconciliation
[Link]("SELECT (SELECT COUNT(*) FROM Source_Staging), (SELECT COUNT(*)
FROM Target_Table)")
src, tgt = [Link]()
status = "PASS" if src == tgt else "FAIL"
details = f"Source: {src}, Target: {tgt}"
log_result("Record Count", status, details)
if status == "FAIL": overall_failure = True
# 2. Duplicate Check
[Link]("SELECT COUNT(*) FROM (SELECT EmployeeID FROM Target_Table
GROUP BY EmployeeID HAVING COUNT(*) > 1) AS D")
dups = [Link]()[0]
status = "PASS" if dups == 0 else "FAIL"
details = f"Found {dups} duplicates"
log_result("Duplicate Check", status, details)
if status == "FAIL": overall_failure = True
# Final Action
if overall_failure:
send_alert("🚨 ETL Validation Failed", f"One or more ETL tests failed. Check {LOG_FILE}
for details.")
print("Results logged. Failures detected.")
else:
print("Validation successful. No issues found.")
[Link]()
except Exception as e:
print(f"Error: {e}")
if __name__ == "__main__":
run_etl_checks()
Key Features of This Script
● Audit Trail: Every run is appended to etl_validation_log.csv. This is great for SLA
reporting and historical trend analysis.
● Proactive Alerts: Instead of manually checking the logs, the script pushes an email
only when something goes wrong.
● Encapsulation: The logic is broken into functions (send_alert, log_result), making it
easy to maintain or expand with more tests.
Comparison of Logging Methods
Feature CSV Logging Console Output Email Alerts
Persistence Permanent record Temporary Immediate
Best For Compliance & Audits Quick Debugging Critical Failures
Effort Low Very Low Moderate (Setup)
Important Security Note: Never hardcode passwords in scripts. For the EMAIL_CONFIG, it is
better to use environment variables or a secret manager.
Would you like me to show you how to schedule this script to run automatically using
Windows Task Scheduler or a Cron job?