基于POC：如何用Python获取Snowflake查询结果并发送至邮箱？

1. 捕获Snowflake插入记录数 #

使用Python Connector for Snowflake执行INSERT语句后，可通过游标对象的rowcount属性直接获取插入的记录数量；如果需要获取插入的具体数据，可在INSERT语句后添加RETURNING *子句，将结果转为DataFrame后统计或处理。

示例代码：

import snowflake.connector
import pandas as pd

# 初始化Snowflake连接
conn = snowflake.connector.connect(
    user='你的用户名',
    password='你的密码',
    account='你的账户标识',
    warehouse='你的仓库',
    database='你的数据库',
    schema='你的Schema'
)

# 执行INSERT语句并捕获结果
cursor = conn.cursor()
try:
    # 基础插入语句
    insert_query = "INSERT INTO target_table(col1, col2) VALUES ('val1', 'val2'), ('val3', 'val4')"
    cursor.execute(insert_query)
    inserted_rows = cursor.rowcount
    print(f"成功插入 {inserted_rows} 条记录")
    
    # 可选：获取插入的详细记录并转为DataFrame
    # cursor.execute("INSERT INTO target_table(col1) VALUES ('test') RETURNING *")
    # df = pd.DataFrame(cursor.fetchall(), columns=[desc[0] for desc in cursor.description])
    # inserted_rows = len(df)
finally:
    cursor.close()
    conn.close()

2. 发送邮件通知 #

使用Python内置的smtplib和email库，将捕获的插入记录数（或DataFrame内容）封装为邮件发送。如果需要展示DataFrame数据，可将其转为HTML表格嵌入邮件正文。

示例代码：

import smtplib
from email.mime.text import MIMEText
from email.header import Header

def send_email(inserted_rows, df=None):
    # 邮件服务器配置
    smtp_server = 'smtp.xxx.com'  # 如smtp.gmail.com、smtp.qq.com
    smtp_port = 587
    sender_email = '你的发件邮箱'
    sender_pwd = '你的邮箱授权码'
    receiver_emails = ['收件人1@xxx.com', '收件人2@xxx.com']
    
    # 构造邮件正文
    content = f"<h3>Snowflake插入操作完成</h3><p>共插入 <strong>{inserted_rows}</strong> 条记录</p>"
    if df is not None:
        # 将DataFrame转为带样式的HTML表格
        html_table = df.to_html(index=False, border=1, classes='table table-bordered')
        content += "<h4>插入记录详情：</h4>" + html_table
    
    # 构建邮件对象
    msg = MIMEText(content, 'html', 'utf-8')
    msg['From'] = Header('Snowflake操作通知', 'utf-8')
    msg['To'] = Header('相关人员', 'utf-8')
    msg['Subject'] = Header(f"Snowflake插入结果通知：{inserted_rows}条记录", 'utf-8')
    
    # 发送邮件
    try:
        server = smtplib.SMTP(smtp_server, smtp_port)
        server.starttls()  # 启用TLS加密
        server.login(sender_email, sender_pwd)
        server.sendmail(sender_email, receiver_emails, msg.as_string())
        print("邮件发送成功")
    except Exception as e:
        print(f"邮件发送失败：{str(e)}")
    finally:
        server.quit()

3. 整合完整流程 #

将连接Snowflake、执行插入、捕获结果、发送邮件的逻辑整合为完整脚本：

import snowflake.connector
import pandas as pd
import smtplib
from email.mime.text import MIMEText
from email.header import Header

def get_snowflake_connection():
    return snowflake.connector.connect(
        user='你的用户名',
        password='你的密码',
        account='你的账户标识',
        warehouse='你的仓库',
        database='你的数据库',
        schema='你的Schema'
    )

def send_email(inserted_rows, df=None):
    smtp_server = 'smtp.xxx.com'
    smtp_port = 587
    sender_email = '你的发件邮箱'
    sender_pwd = '你的邮箱授权码'
    receiver_emails = ['收件人@xxx.com']
    
    content = f"<h3>Snowflake POC插入操作完成</h3><p>共插入 <strong>{inserted_rows}</strong> 条记录</p>"
    if df is not None:
        content += "<h4>插入记录详情：</h4>" + df.to_html(index=False, border=1)
    
    msg = MIMEText(content, 'html', 'utf-8')
    msg['From'] = Header('Snowflake POC通知', 'utf-8')
    msg['To'] = Header('相关人员', 'utf-8')
    msg['Subject'] = Header(f"Snowflake插入结果：{inserted_rows}条记录", 'utf-8')
    
    try:
        server = smtplib.SMTP(smtp_server, smtp_port)
        server.starttls()
        server.login(sender_email, sender_pwd)
        server.sendmail(sender_email, receiver_emails, msg.as_string())
        print("通知邮件发送成功")
    except Exception as e:
        print(f"邮件发送失败：{str(e)}")
    finally:
        server.quit()

if __name__ == "__main__":
    conn = None
    cursor = None
    try:
        conn = get_snowflake_connection()
        cursor = conn.cursor()
        
        # 执行INSERT语句
        insert_query = "INSERT INTO test_table(name, age) VALUES ('张三', 25), ('李四', 30)"
        cursor.execute(insert_query)
        inserted_rows = cursor.rowcount
        
        # 可选：获取插入的详细记录
        # cursor.execute("SELECT * FROM test_table WHERE name IN ('张三', '李四')")
        # df = pd.DataFrame(cursor.fetchall(), columns=[desc[0] for desc in cursor.description])
        
        # 发送通知邮件
        send_email(inserted_rows)  # 若有df则改为send_email(inserted_rows, df)
        
    except Exception as e:
        print(f"操作失败：{str(e)}")
        # 可选：发送失败通知
        # send_email(0)
    finally:
        if cursor:
            cursor.close()
        if conn:
            conn.close()

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