# server: if 1: from fastapi import FastAPI from apscheduler.schedulers.background import BackgroundScheduler import time app FastAPI() scheduler BackgroundScheduler(timezoneAsia/Shanghai) is_running False # 任务锁防止上一轮没跑完下一轮又触发 def run_sql_task(): global is_running if is_running: print(f[{time.ctime()}] 任务正在执行跳过本次触发) return try: is_running True print(f 定时任务开始 {time.strftime(%Y-%m-%d %H:%M:%S)} ) # # 这里放你的业务代码遍历sqlite查询unit_data_his去重处理 # print(f 定时任务结束 {time.strftime(%Y-%m-%d %H:%M:%S)} ) except Exception as e: print(f任务异常{e}) finally: is_running False # cron配置每小时的55分执行 scheduler.add_job( run_sql_task, triggercron, minute*, # 每小时55分触发 idhour_55_sql_task, replace_existingTrue ) scheduler.start() print(APScheduler已启动每小时55分执行任务) app.get(/) def root(): return {msg: 服务正常定时任务每小时55分运行} import uvicorn uvicorn.run(app, host0.0.0.0, port8000) fastapi 定时任务
# server: if 1: from fastapi import FastAPI from apscheduler.schedulers.background import BackgroundScheduler import time app FastAPI() scheduler BackgroundScheduler(timezoneAsia/Shanghai) is_running False # 任务锁防止上一轮没跑完下一轮又触发 def run_sql_task(): global is_running if is_running: print(f[{time.ctime()}] 任务正在执行跳过本次触发) return try: is_running True print(f 定时任务开始 {time.strftime(%Y-%m-%d %H:%M:%S)} ) # # 这里放你的业务代码遍历sqlite查询unit_data_his去重处理 # print(f 定时任务结束 {time.strftime(%Y-%m-%d %H:%M:%S)} ) except Exception as e: print(f任务异常{e}) finally: is_running False # cron配置每小时的55分执行 scheduler.add_job( run_sql_task, triggercron, minute*, # 每小时55分触发 idhour_55_sql_task, replace_existingTrue ) scheduler.start() print(APScheduler已启动每小时55分执行任务) app.get(/) def root(): return {msg: 服务正常定时任务每小时55分运行} import uvicorn uvicorn.run(app, host0.0.0.0, port8000)