FastAPI 使用apscheduler轻松实现定时任务

说到定时任务(corntab)大家肯定都不陌生，但在一个高频请求web的项目中，有一个方便、好用的定时任务，绝对可以省心又省力。 今天说到的就是我多次提到的apscheduler这个库来实现的，也是我目前关于所有定时任务使用最常用的一个框架。

背景

之前我使用的不是很规范，所以在startup中直接启用了定时任务，其实也能用，但是存在不够优雅，和FastAPI本身没有整合，但因为能用，我也就懒得折腾。 最近在做新项目，就想着把这块用好的方法维护起来，就有这篇文章了。

整合到项目中

遵循最小原则，先看项目的结构如下

my_fastapi_project/
├── main.py          # 应用入口
├── api/
│   ├── __init__.py
│   └── routes.py    # API路由定义
├── tasks/
│   ├── __init__.py
│   └── scheduled_tasks.py  # 定时任务定义
├── core/
│   ├── __init__.py
│   └── scheduler.py # 调度器初始化和配置
└── requirements.txt

# core/scheduler.py
# 主要是引入支持异步的定时框架
from apscheduler.schedulers.asyncio import AsyncIOScheduler

scheduler = AsyncIOScheduler()

# tasks/scheduled_tasks.py
from apscheduler.triggers.interval import IntervalTrigger
from apscheduler.triggers.cron import CronTrigger
from datetime import datetime
from core.scheduler import scheduler

# 每隔1分钟执行一次的任务
@scheduler.scheduled_job(IntervalTrigger(minutes=1))
async def interval_task():
    print(f"Interval Task - 执行时间: {datetime.now()}")

# 每天凌晨0点执行的任务
@scheduler.scheduled_job(CronTrigger(hour=0, minute=0))
async def daily_task():
    print(f"Daily Task - 执行时间: {datetime.now()}")

# 一次性任务，在指定时间执行
@scheduler.scheduled_job('date', run_date='2023-12-31 23:59:59')
async def one_time_task():
    print(f"One-Time Task - 执行时间: {datetime.now()}")

# main.py
from fastapi import FastAPI
from core.scheduler import scheduler
from api.routes import api_router

app = FastAPI()

# 包含API路由
app.include_router(api_router)

# 在应用启动时启动调度器
@app.on_event("startup")
async def startup_event():
    scheduler.start()
    print("Scheduler started")

# 在应用关闭时关闭调度器
@app.on_event("shutdown")
async def shutdown_event():
    scheduler.shutdown()
    print("Scheduler shutdown")

# api/routes.py

from fastapi import APIRouter
from datetime import datetime
from core.scheduler import scheduler
router = APIRouter()

@router.get("/")
async def read_root():
   return {"status": "running" if scheduler.running else "stopped"}

总结

上面的代码就已经实现了一个最基本的定时任务,可以在改写成各种使用场景，这样有个好处就是可以进行模块化管理。

将调度器初始化、任务定义和应用启动分别放在不同的模块中，便于管理和扩展。 通过 Python 的模块导入机制，可以在不同文件之间共享调度器实例和任务函数。 任务类型： IntervalTrigger：按固定间隔执行任务。 CronTrigger：按cron表达式指定的时间执行任务，适合周期性任务。 DateTrigger：在指定的未来时间点执行任务一次。 应用生命周期管理： 使用 FastAPI 的 startup 和 shutdown 事件来管理调度器的启动和关闭，确保资源的正确管理。