如何用Python实现当月月末自动从Snowflake提取数据？

一、完善月末日期判断逻辑 #

你原有的月末计算代码在12月会出现错误（month + 1 = 13超出合法月份范围），这里提供两种更可靠的实现方式：

方式1：使用calendar模块（推荐）

from datetime import datetime
import calendar

def is_today_end_of_month():
    today = datetime.today()
    # 获取当月最后一天的日期
    _, last_day = calendar.monthrange(today.year, today.month)
    # 判断当日是否为月末
    return today.day == last_day

方式2：修复原代码的12月边界问题

from datetime import datetime, timedelta

def get_end_of_month():
    today = datetime.today()
    if today.month == 12:
        # 12月的下一月是次年1月，减1天得到12月最后一天
        end_of_month = datetime(today.year + 1, 1, 1) - timedelta(days=1)
    else:
        end_of_month = datetime(today.year, today.month + 1, 1) - timedelta(days=1)
    return end_of_month

def is_today_end_of_month():
    # 仅比较日期部分，忽略时分秒差异
    return datetime.today().date() == get_end_of_month().date()

二、集成Snowflake数据提取功能 #

首先安装Snowflake Python连接器：

pip install snowflake-connector-python

编写数据提取与保存的函数：

import snowflake.connector
from datetime import datetime

def extract_data_from_snowflake():
    # 替换为你的Snowflake账户信息
    conn_config = {
        "user": "你的用户名",
        "password": "你的密码",
        "account": "你的账户标识（格式：<账户名>.<区域>，如abc123.us-west-2）",
        "warehouse": "你的仓库名",
        "database": "你的数据库名",
        "schema": "你的模式名"
    }

    conn = None
    cursor = None
    try:
        # 建立连接
        conn = snowflake.connector.connect(**conn_config)
        cursor = conn.cursor()

        # 替换为你的实际查询语句
        query = "SELECT * FROM 你的目标表名;"
        cursor.execute(query)

        # 获取查询结果
        results = cursor.fetchall()
        # 获取表头信息
        headers = [col[0] for col in cursor.description]

        # 将数据保存为CSV文件（文件名包含日期）
        output_filename = f"snowflake_extracted_data_{datetime.today().strftime('%Y%m%d')}.csv"
        with open(output_filename, "w", newline="", encoding="utf-8") as f:
            import csv
            writer = csv.writer(f)
            writer.writerow(headers)
            writer.writerows(results)

        print(f"数据已成功提取并保存至：{output_filename}")
    except Exception as e:
        print(f"数据提取失败：{str(e)}")
    finally:
        # 确保资源被释放
        if cursor:
            cursor.close()
        if conn:
            conn.close()

三、整合逻辑，触发自动提取 #

将月末判断与数据提取逻辑整合，运行脚本时自动检查并执行：

if __name__ == "__main__":
    if is_today_end_of_month():
        print("检测到今日为月末，开始执行Snowflake数据提取...")
        extract_data_from_snowflake()
    else:
        print("今日不是月末，无需执行数据提取")

四、定时运行建议 #

要实现每天自动检查运行，可借助系统定时任务工具：

• Windows：使用「任务计划程序」，创建每日定时任务运行该Python脚本
• Linux/macOS：使用crontab，添加定时规则（例如每天凌晨2点运行：0 2 * * * python3 /path/to/your/script.py）

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