当前位置：首页>python>python自动化系列:Python + schedule库轻松实现每日定时数据库操作

python自动化系列:Python + schedule库轻松实现每日定时数据库操作

2026-04-16 03:42:21

作品介绍

作品名称：Python + schedule库轻松实现每日定时数据库操作

用到的库：pymysql、time、datetime、schedule

作品简介：该实例实现了一个定时任务，该任务每天晚上11点自动更新指定数据库表中的记录。

实现过程

一、导入必要的库

import pymysqlimport timefrom datetime import datetime, timedeltaimport schedule

二、定义函数update_record

使用pymysql.connect()创建数据库连接，并使用上下文管理器确保连接正确关闭。
构建并执行一条UPDATE SQL语句来更新指定记录。
提交更改到数据库，并打印一条包含当前时间和受影响表名的日志信息。
如果发生异常（例如数据库连接失败或SQL执行错误），捕获异常并打印错误信息。

def update_record(db, table, column, value, condition):    try:        with pymysql.connect(host='localhost', user='root', password='', db=db) as conn:            with conn.cursor() as cursor:                sql = f"UPDATE {table} SET {column} = %s WHERE {condition}"                cursor.execute(sql, (value,))                conn.commit()                print(f"{datetime.now()}: Updated record in {table}")    except Exception as e:        print(f"Failed to update record: {e}")

三、定义函数setup_daily_update

使用schedule.every().day.at("23:00").do(...)安排update_record函数在每天23:00时运行，传入所需的参数。

def setup_daily_update():    # 设置每天晚上11点执行的任务    schedule.every().day.at("23:00").do(update_record, "mydb", "users", "status", "active", "id = 1")

四、主程序逻辑

初始化：

调用setup_daily_update()函数配置定时任务。
打印一条消息表示调度器已启动，并等待下一次预定任务的到来。

无限循环：

使用while True:创建一个无限循环，使程序持续运行。
在每次迭代中调用schedule.run_pending()检查是否有任何待执行的任务。
然后调用time.sleep(1)让程序暂停1秒钟，减少CPU占用，同时保持对定时任务的响应性。

if __name__ == "__main__":    setup_daily_update()    print("Scheduler started. Waiting for the next scheduled task...")    while True:        schedule.run_pending()        time.sleep(1)  # 等待1秒后再检查是否有待执行的任务