如何在FastAPI中仅工作时段于应用空闲时运行后台任务

完全可以实现需求，核心思路是定时触发任务 + 时段校验 + 空闲状态检测，下面提供两种实用方案，分别基于你提到的@repeat_every和更灵活的APScheduler。

方案一：使用fastapi-utils的@repeat_every #

@repeat_every是fastapi-utils库提供的装饰器，能在应用启动后自动启动后台线程定期触发任务，无需依赖外部请求。

步骤1：安装依赖 #

pip install fastapi-utils

步骤2：实现代码 #

from fastapi import FastAPI, Request
from fastapi_utils.tasks import repeat_every
from datetime import datetime
from threading import Lock

app = FastAPI()

# 线程安全的请求计数器，用于判断应用是否空闲
request_counter = 0
counter_lock = Lock()

# 中间件：统计当前正在处理的请求数
@app.middleware("http")
async def count_active_requests(request: Request, call_next):
    global request_counter
    with counter_lock:
        request_counter += 1
    response = await call_next(request)
    with counter_lock:
        request_counter -= 1
    return response

# 启动时注册重复任务，每分钟触发一次检查
@app.on_event("startup")
@repeat_every(seconds=60)  # 可根据需求调整检查间隔
def run_idle_background_task():
    now = datetime.now()
    # 1. 校验是否在9:00-21:00时段内
    if 9 <= now.hour < 21:
        # 2. 校验应用是否空闲（无正在处理的请求）
        with counter_lock:
            if request_counter == 0:
                # 3. 执行你的后台任务逻辑
                print(f"执行后台任务 | 时间: {now.strftime('%Y-%m-%d %H:%M:%S')}")
                # 任务完成后无需手动重复，@repeat_every会自动按间隔触发下一次检查

方案二：使用APScheduler（更灵活的定时控制） #

APScheduler支持更精细的定时规则（如Cron表达式），适合需要精准时段触发的场景。

步骤1：安装依赖 #

pip install apscheduler

步骤2：实现代码 #

from fastapi import FastAPI, Request
from apscheduler.schedulers.background import BackgroundScheduler
from datetime import datetime
from threading import Lock

app = FastAPI()

request_counter = 0
counter_lock = Lock()

@app.middleware("http")
async def count_active_requests(request: Request, call_next):
    global request_counter
    with counter_lock:
        request_counter += 1
    response = await call_next(request)
    with counter_lock:
        request_counter -= 1
    return response

def idle_background_task():
    # 仅检查空闲状态（时段控制交给Cron表达式）
    with counter_lock:
        if request_counter == 0:
            print(f"执行后台任务 | 时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")

@app.on_event("startup")
def start_scheduler():
    scheduler = BackgroundScheduler()
    # 使用Cron表达式限定触发时段：每天9:00-20:59，每分钟触发一次
    scheduler.add_job(
        idle_background_task,
        'cron',
        hour='9-20',
        minute='*'
    )
    scheduler.start()

@app.on_event("shutdown")
def shutdown_scheduler():
    scheduler.shutdown()

关键注意事项 #

1. 空闲状态判断：通过中间件统计当前活跃请求数，确保仅在无请求处理时执行任务，避免占用应用资源。
2. 线程/协程安全：如果你的应用以异步路由为主（使用async def），请将threading.Lock替换为asyncio.Lock，并调整计数器操作逻辑为异步方式。
3. 任务轻量化：后台任务如果耗时较长，建议放在单独的线程/进程中执行，避免阻塞FastAPI的事件循环或线程池。

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