如何用Python每小时自动从SQL表获取前一小时数据

首先，我们需要基于当前运行时间计算整点对齐的时间范围——不管脚本什么时候触发，都取当前所在小时的整点作为结束时间，前推一小时作为起始时间。这样即使脚本延迟运行，也能保证拉取的是正确的整小时数据。

具体实现步骤 #

1. 引入datetime模块处理时间计算
2. 计算对齐到整点的结束时间（比如16:00运行或16:02延迟运行，结束时间都是16:00）
3. 计算起始时间为结束时间前推1小时（即15:00）
4. 将时间格式化为SQL兼容的字符串（比如YYYY-MM-DD HH:MM:SS）
5. 使用参数化查询避免SQL注入风险（比直接拼接字符串更安全）

修改后的完整代码 #

import pandas as pd
from datetime import datetime, timedelta

def CustomerData():
    # 获取当前时间，对齐到整点作为结束时间
    now = datetime.now()
    end_time = now.replace(minute=0, second=0, microsecond=0)
    # 起始时间为结束时间前推1小时
    start_time = end_time - timedelta(hours=1)
    
    # 格式化为SQL兼容的字符串（若数据库用UTC时间，替换为datetime.utcnow()）
    start_str = start_time.strftime('%Y-%m-%d %H:%M:%S')
    end_str = end_time.strftime('%Y-%m-%d %H:%M:%S')
    
    # 参数化查询，避免SQL注入
    sql = """
        SELECT Name, CustomerID, PurchaseDateTime 
        FROM XYZ 
        WHERE PurchaseDateTime BETWEEN %s AND %s
    """
    DF = pd.read_sql(sql, engine, params=(start_str, end_str))
    DF.columns = ['Name', 'CustomerID', 'PurchaseDate']
    return DF

关键细节说明 #

• 整点对齐逻辑：now.replace(minute=0, second=0, microsecond=0)会把当前时间的分、秒、微秒清零，得到当前小时的整点时间。比如16:02运行时，end_time就是16:00，start_time就是15:00。
• 时区处理：如果数据库存储的是UTC时间，记得把datetime.now()换成datetime.utcnow()，避免时区不一致导致的数据偏差。
• 参数化查询：用params传递时间参数比直接拼接字符串更安全，能有效防止SQL注入攻击，同时也避免了时间格式拼接的潜在错误。

额外优化建议 #

如果你的脚本是通过定时任务（比如Linux的cron、Windows的任务计划）触发的，也可以让定时任务严格按整点执行（比如每小时的0分0秒触发），这样双重保障时间范围的准确性。不过即使定时任务偶尔延迟，上面的代码依然能保证拉取正确的整小时数据。

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