Python基于Fastapi与APScheduler的应用定时任务

APScheduler==3.11.0
fastapi==0.119.0
uvicorn==0.37.0

from contextlib import asynccontextmanager
from datetime import datetime
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.triggers.interval import IntervalTrigger
from fastapi import FastAPI
from logger import logger

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 初始化调度器
    scheduler = AsyncIOScheduler()
    # 添加定时任务，每分钟执行一次
    scheduler.add_job(
        feedback_task,
        trigger=IntervalTrigger(minutes=1),
        id="feedback_task",
        replace_existing=True
    )
    # 启动调度器
    scheduler.start()
    logger.info("APScheduler定时器已启动")
    # 应用运行期间
    yield
    # 关闭逻辑
    scheduler.shutdown()
    logger.info("APScheduler定时器已关闭")

async def feedback_task():
    logger.info(f"feedback_task start : {datetime.now()}")

import uvicorn

from fastapi import FastAPI
from route import HealthRouter
from utils.ScheduleTask import lifespan

app = FastAPI(lifespan=lifespan)
app.include_router(HealthRouter.router)

if __name__ == '__main__':
    # 启动服务
    uvicorn.run(app, host="0.0.0.0", port=8000)