如何用Python FASTAPI实现每日15:30仅执行一次的定时任务？

你的代码问题分析 #

你当前代码的核心问题是误用了schedule.run_all()：

• 这个方法会立即执行所有已注册的定时任务，而非等待指定时间触发
• 配合while True死循环，会导致任务被反复执行，完全不符合“每日仅执行一次”的需求

方法一：修正schedule库的用法 #

把schedule.run_all()换成schedule.run_pending()，它只会检查并执行到点的任务，同时添加time.sleep()降低CPU占用：

from fastapi import FastAPI
import schedule
import time
import threading

app = FastAPI()

def job2():
    print("每日15:30执行的任务")

# 调度每日15:30的任务
schedule.every().day.at("15:30:00").do(job2)

def run_schedule():
    while True:
        schedule.run_pending()
        time.sleep(60)  # 每分钟检查一次，可根据需求调整间隔

# 启动单独线程运行定时任务，避免阻塞FastAPI主线程
threading.Thread(target=run_schedule, daemon=True).start()

方法二：用APScheduler适配FastAPI异步环境 #

因为FastAPI是异步框架，更推荐使用APScheduler，它支持异步任务调度，且能更好地集成到FastAPI的生命周期中：

1. 先安装依赖：

pip install apscheduler

2. 编写代码：

from fastapi import FastAPI
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from datetime import datetime

app = FastAPI()
scheduler = AsyncIOScheduler(timezone="Asia/Shanghai")  # 替换为你的时区

async def job2():
    print(f"任务执行于{datetime.now()}")

# 添加每日15:30的定时任务
scheduler.add_job(job2, "cron", hour=15, minute=30)

# FastAPI启动时启动调度器
@app.on_event("startup")
async def startup_event():
    scheduler.start()

# FastAPI关闭时停止调度器
@app.on_event("shutdown")
async def shutdown_event():
    scheduler.shutdown()

注意：如果你的任务是同步的，也可以用BlockingScheduler，但需要放在单独线程中运行，避免阻塞FastAPI主线程。

内容的提问来源于stack exchange，提问作者Arzu Yıldız