FastAPI中如何用Lifespan Events实现@repeat_every定时任务

问题根源 #

原来用@app.on_event("startup")配合@repeat_every的写法被弃用后，直接把任务放进Lifespan上下文只会在启动时执行一次——因为Lifespan的启动逻辑（yield前的代码）仅运行一次。给Lifespan函数加@repeat_every会触发参数错误，是因为Lifespan必须接收app参数，但装饰器修改了函数签名，导致调用时传参不匹配。

可行实现方案 #

方案1：手动后台线程+循环（无额外依赖） #

用Python内置的threading模块启动后台线程，在线程内循环执行任务，同时通过事件控制线程停止：

from fastapi import FastAPI
from contextlib import asynccontextmanager
import threading
import time

# 你的定时任务函数
def saveToReports():
    # 这里写你的报表保存逻辑
    print("执行报表保存任务")

# 用于终止线程的信号
stop_task = threading.Event()

def scheduled_task_loop():
    while not stop_task.is_set():
        saveToReports()
        # 等待30分钟，若期间收到停止信号则提前退出
        stop_task.wait(30 * 60)

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 启动后台线程，daemon=True确保退出时线程自动终止
    task_thread = threading.Thread(target=scheduled_task_loop, daemon=True)
    task_thread.start()
    yield  # 应用运行期间停在这里
    # 关闭时发送停止信号，等待线程结束
    stop_task.set()
    task_thread.join()

app = FastAPI(lifespan=lifespan)

方案2：用APScheduler（专业定时任务库） #

如果需要更复杂的定时规则（比如按 cron 表达式执行），推荐用APScheduler，功能更完善：

1. 先安装依赖：

pip install apscheduler

2. 代码实现：

from fastapi import FastAPI
from contextlib import asynccontextmanager
from apscheduler.schedulers.background import BackgroundScheduler

def saveToReports():
    print("执行报表保存任务")

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 初始化后台调度器
    scheduler = BackgroundScheduler()
    # 添加任务：每30分钟执行一次
    scheduler.add_job(saveToReports, 'interval', minutes=30)
    # 启动调度器
    scheduler.start()
    yield
    # 关闭调度器，等待任务完成后退出
    scheduler.shutdown(wait=True)

app = FastAPI(lifespan=lifespan)

注意事项 #

• 方案1的daemon=True可以确保当FastAPI进程退出时，后台线程自动终止，避免僵尸线程。
• APScheduler支持多种触发器（interval、cron、date），可以根据需求灵活配置。

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