如何在FastAPI中用AsyncIOScheduler每日执行带起止时间的调度任务

你的核心问题是启动时固定的start/end_time无法每日更新，因为lifespan仅在应用启动时执行一次。要实现每日6点启动、持续到8点的间隔任务，有两种可靠方案：

方案一：每日6点动态创建当日的间隔任务 #

通过cron任务触发，每天6点时创建一个仅在当日6:00-8:00运行的间隔任务，到时间后任务会自动停止，无需手动清理。

修改后的代码：

@asynccontextmanager
async def lifespan(app: FastAPI):
    from apscheduler.schedulers.asyncio import AsyncIOScheduler
    from datetime import datetime, timedelta

    def print_hello():
        print("Hello", datetime.now())

    def create_daily_interval_job():
        # 计算当日的6点和8点
        today = datetime.now().date()
        start_time = datetime(today.year, today.month, today.day, 6, 0, 0)
        end_time = datetime(today.year, today.month, today.day, 8, 0, 0)
        
        # 添加当日的间隔任务，每秒执行一次，到8点自动停止
        scheduler.add_job(
            print_hello, 
            "interval", 
            seconds=1, 
            start_date=start_time,
            end_date=end_time,
            id="daily_interval_job",  # 给任务加ID，避免重复创建
            replace_existing=True
        )

    scheduler = AsyncIOScheduler()
    
    # 每天6点触发创建当日的间隔任务
    scheduler.add_job(
        create_daily_interval_job, 
        'cron', 
        hour=6, 
        minute=0,
        id="create_daily_job"
    )

    # 首次启动时如果当前时间在6-8点，立即创建任务
    now = datetime.now()
    today_start = datetime(now.year, now.month, now.day, 6, 0, 0)
    today_end = datetime(now.year, now.month, now.day, 8, 0, 0)
    if today_start <= now <= today_end:
        create_daily_interval_job()

    scheduler.start()
    
    yield
    scheduler.shutdown()

方案二：直接让间隔任务仅在每日6-8点运行 #

无需动态创建任务，直接给间隔任务设置时间范围，或在任务内部判断当前时间是否符合要求：

方式A：任务内部判断时间区间 #

@asynccontextmanager
async def lifespan(app: FastAPI):
    from apscheduler.schedulers.asyncio import AsyncIOScheduler
    from datetime import datetime

    def print_hello():
        now = datetime.now()
        # 仅在6-8点执行逻辑
        if 6 <= now.hour < 8:
            print("Hello", now)

    scheduler = AsyncIOScheduler()
    
    # 每秒执行一次任务，内部判断时间是否符合要求
    scheduler.add_job(print_hello, "interval", seconds=1)

    scheduler.start()
    
    yield
    scheduler.shutdown()

方式B：用cron直接约束运行窗口 #

利用cron表达式指定任务仅在每日6:00-7:59:59之间每秒执行：

@asynccontextmanager
async def lifespan(app: FastAPI):
    from apscheduler.schedulers.asyncio import AsyncIOScheduler
    from datetime import datetime

    def print_hello():
        print("Hello", datetime.now())

    scheduler = AsyncIOScheduler()
    
    # 每天6:00-7:59:59之间，每秒执行一次
    scheduler.add_job(
        print_hello, 
        "cron", 
        hour="6-7",
        minute="*",
        second="*"
    )

    scheduler.start()
    
    yield
    scheduler.shutdown()

关键说明 #

• 方案一适合严格要求6点准时启动、8点准时停止的场景，任务仅在当日指定时间段内存在，资源占用更低。
• 方案二的方式A逻辑简单，但会每秒触发一次任务（即使不在目标时间），适合对性能要求不高的场景；方式B用cron直接约束时间，更高效，无需内部判断。

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