如何用Python在Snowflake创建含%tbl%的表时发送邮件通知

前置准备 #

1. 安装依赖包：

pip install snowflake-connector-python python-dotenv

2. 创建.env文件存储敏感配置（避免硬编码）：

# Snowflake配置
SNOWFLAKE_ACCOUNT=你的账号
SNOWFLAKE_USER=你的用户名
SNOWFLAKE_PASSWORD=你的密码
SNOWFLAKE_WH=你的仓库名
SNOWFLAKE_DB=要监控的数据库
SNOWFLAKE_SCHEMA=要监控的模式

# 邮件配置
SMTP_SERVER=smtp.qq.com  # 替换为你的SMTP服务器
SMTP_PORT=465
SENDER_EMAIL=你的发件邮箱
SENDER_PASSWORD=邮箱授权码  # 注意：不是登录密码，是邮箱的SMTP授权码
RECEIVER_EMAIL=收件人邮箱

完整示例代码 #

import os
import datetime
from dotenv import load_dotenv
import snowflake.connector
from snowflake.connector.errors import ProgrammingError
import smtplib
from email.mime.text import MIMEText

# 加载环境变量
load_dotenv()

# 配置常量
LAST_CHECK_FILE = "last_check_time.txt"
TARGET_TABLE_KEYWORD = "%tbl%"

def get_last_check_time():
    """读取上次检查的时间，不存在则返回当前时间的前1小时"""
    if os.path.exists(LAST_CHECK_FILE):
        with open(LAST_CHECK_FILE, "r") as f:
            return datetime.datetime.fromisoformat(f.read().strip())
    else:
        return datetime.datetime.now() - datetime.timedelta(hours=1)

def update_last_check_time():
    """更新上次检查时间为当前时间"""
    with open(LAST_CHECK_FILE, "w") as f:
        f.write(datetime.datetime.now().isoformat())

def query_new_tables(last_check_time):
    """查询Snowflake中指定时间后创建的、名称包含关键词的表"""
    conn = None
    try:
        # 建立Snowflake连接
        conn = snowflake.connector.connect(
            account=os.getenv("SNOWFLAKE_ACCOUNT"),
            user=os.getenv("SNOWFLAKE_USER"),
            password=os.getenv("SNOWFLAKE_PASSWORD"),
            warehouse=os.getenv("SNOWFLAKE_WH"),
            database=os.getenv("SNOWFLAKE_DB"),
            schema=os.getenv("SNOWFLAKE_SCHEMA")
        )
        
        # 执行查询
        cursor = conn.cursor()
        query = f"""
            SELECT TABLE_NAME, CREATED
            FROM INFORMATION_SCHEMA.TABLES
            WHERE TABLE_TYPE = 'BASE TABLE'
            AND TABLE_NAME LIKE '{TARGET_TABLE_KEYWORD}'
            AND CREATED > '{last_check_time.isoformat()}'
            ORDER BY CREATED DESC
        """
        cursor.execute(query)
        return cursor.fetchall()
    
    except ProgrammingError as e:
        print(f"查询Snowflake出错: {e}")
        return []
    finally:
        if conn:
            conn.close()

def send_email_notification(tables):
    """发送邮件通知"""
    if not tables:
        return
    
    # 构造邮件内容
    table_list = "\n".join([f"- 表名: {table[0]}, 创建时间: {table[1]}" for table in tables])
    subject = f"【Snowflake通知】检测到新创建的含关键词表"
    body = f"以下是最近创建的名称包含'{TARGET_TABLE_KEYWORD}'的表:\n\n{table_list}"
    
    msg = MIMEText(body, "plain", "utf-8")
    msg["Subject"] = subject
    msg["From"] = os.getenv("SENDER_EMAIL")
    msg["To"] = os.getenv("RECEIVER_EMAIL")
    
    try:
        # 发送邮件
        with smtplib.SMTP_SSL(os.getenv("SMTP_SERVER"), int(os.getenv("SMTP_PORT"))) as server:
            server.login(os.getenv("SENDER_EMAIL"), os.getenv("SENDER_PASSWORD"))
            server.send_message(msg)
        print("邮件通知发送成功")
    except Exception as e:
        print(f"发送邮件出错: {e}")

if __name__ == "__main__":
    # 获取上次检查时间
    last_time = get_last_check_time()
    # 查询新表
    new_tables = query_new_tables(last_time)
    # 发送通知
    send_email_notification(new_tables)
    # 更新检查时间
    update_last_check_time()

关键说明 #

• 元数据查询逻辑：通过INFORMATION_SCHEMA.TABLES视图获取表的创建时间和名称，这个视图是Snowflake内置的系统视图，包含了当前数据库模式下的所有表信息。
• 重复通知避免：用本地文件last_check_time.txt记录上次检查的时间，每次查询只获取该时间点之后创建的表，避免重复发送通知。
• 邮件配置注意：大多数邮箱需要开启SMTP服务并使用授权码（而非登录密码）进行验证，比如QQ邮箱需要在设置中开启"SMTP服务"并生成授权码。
• 定时执行：如果需要持续监控，可以用操作系统的定时任务（如Linux的cron、Windows的任务计划程序）定期运行这个脚本，比如每5分钟执行一次。

参考提示 #

• Snowflake官方文档中关于INFORMATION_SCHEMA.TABLES视图的详细字段说明
• Python官方文档中smtplib和email模块的使用指南

内容的提问来源于stack exchange，提问作者AJS