FastAPI中重复任务定时器修饰器与路由混搭导致日志错乱问题的解决

阅读提示：本文共计约5739个文字，预计阅读时间需要大约15分钟，由作者韩宝贝bj编辑整理创作于2023年11月05日23时57分15秒。

在FastAPI项目中，有时我们需要使用repeat_task定时任务修饰器来实现周期性的任务执行，同时还需要通过路由来处理HTTP请求。然而，在某些情况下，这两种功能可能会产生冲突，导致日志写入错乱。下面是一个简单的解决方案，帮助您在使用repeat_task定时任务修饰器的同时，确保日志的正确写入。

导入所需的库和模块：

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.middleware.cors import CORSMiddleware
import uvicorn
from starlette.middleware.sessions import SessionMiddleware
from starlette.responses import RedirectResponse
from starlette.staticfiles import StaticFiles
from starlette.templating import Jinja2Templates
from starlette.websockets import WebSocketState
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.sql.functions import current_timestamp
from datetime import datetime
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.triggers.interval import IntervalTrigger
from apscheduler.jobstores.asyncio import AsyncIODiskJobStore
from apscheduler.executors.asyncio import AsyncIOExecutor
from logger import get_logger
from models import User, db
from routes import router as api_router

接下来，定义一个全局变量，用于存储当前时间戳，以便在重复任务触发时进行区分：

current_time = None

然后，创建一个定时任务装饰器函数，用于在重复任务触发时记录当前时间戳：

def repeat_task(func):
    @wraps(func)
    async def wrapper(*args, **kwargs):
        global current_time
        current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        return await func(*args, **kwargs)
    return wrapper

现在，您可以像往常一样使用repeat_task定时任务装饰器，例如：

@app.on_event("startup")
async def startup_event():
    scheduler = AsyncIOScheduler(jobstores=[AsyncIODiskJobStore()], executors=[AsyncIOExecutor()])
    scheduler.add_job(func=my_function, trigger=IntervalTrigger(seconds=60), id="my_job")
    scheduler.start()

@repeat_task
async def my_function():
    # 在这里执行您的重复任务...
    get_logger().info(f"Repeating task triggered at {current_time}")

确保在项目的入口点（例如main.py）中正确配置FastAPI实例、路由和其他中间件，例如：

app = FastAPI(
    title="My Application",
    description="A simple description of my application",
    version="0.1.0",
    debug=True,
    templates=Jinja2Templates(directory="templates"),
    middleware=[
        SessionMiddleware,
        CORSMiddleware,
    ],
)

app.include_router(api_router, tags=["API"], prefix="/api/v1")

if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=8000, reload=True)

这样，当您在FastAPI应用程序中使用repeat_task定时任务修饰器时，可以确保日志的正确写入，避免与其他功能产生冲突。