FastAPI中asynccontextmanager修饰的lifespan函数未执行,无法启动AsyncIOScheduler定时任务
FastAPI中asynccontextmanager修饰的lifespan函数未执行,无法启动AsyncIOScheduler定时任务
我太懂你这种抓瞎的感觉了——明明写了启动逻辑,结果连个影子都看不到,定时任务自然也泡汤了。咱们直接戳核心问题,你的代码里有两个关键错误,导致FastAPI完全没调用你的lifespan函数:
问题1:lifespan函数没关联到FastAPI实例
你只是定义了lifespan函数,但从来没告诉FastAPI要使用它。FastAPI不会自动扫描并使用你定义的lifespan,必须在初始化FastAPI对象时显式传入这个参数。
问题2:代码顺序完全反了
你先创建了app = FastAPI(),之后才定义lifespan函数。这就相当于先盖房子再画设计图——FastAPI初始化的时候,lifespan还不存在,就算你想传参数也传不了。
解决步骤
1. 调整代码顺序 + 关联lifespan
把lifespan函数的定义移到app = FastAPI()之前,并且在初始化时传入lifespan参数:
# 所有导入代码不变... dictConfig(log_config) app_logger = logging.getLogger("app") load_dotenv(".env") # 第一步:先定义lifespan函数 @asynccontextmanager async def lifespan(app: FastAPI): # type: ignore print("************ LIFESPAN FUNCTION STARTED ************") try: scheduler = AsyncIOScheduler() # Monday to Friday at 05:00 trigger = CronTrigger( year="*", month="*", day="*", day_of_week="mon-fri", hour="5", minute="0", second="0", ) scheduler.add_job(update_ohlc_rsi_charts_for_tickers, trigger) scheduler.start() log_msg = "Inside lifespan: " + str(scheduler.get_jobs()) print(log_msg) app_logger.info(log_msg) yield except Exception as e: print(f"************ EXCEPTION IN LIFESPAN: {e} ************") import traceback traceback.print_exc() # 如果你想让FastAPI启动失败可以取消下面的注释 # raise e finally: print("************ LIFESPAN FUNCTION SHUTDOWN INITIATED ************") if "scheduler" in locals() and scheduler.running: scheduler.shutdown() print("************ LIFESPAN FUNCTION SHUTDOWN COMPLETE ************") # 第二步:创建FastAPI实例时,传入lifespan参数 app = FastAPI(lifespan=lifespan) # 最后定义路由 @app.get("/") async def root() -> dict: ticker = "GLD" log_msg = f"Inside root: {ticker=}" app_logger.info(log_msg) return { "message": f"Hello World RSI, {ticker=}", }
2. 验证效果
修改后重新启动uvicorn(不管用不用--reload都可以),你应该能在日志里看到:
************ LIFESPAN FUNCTION STARTED ************ Inside lifespan: [<Job (id=... name=update_ohlc_rsi_charts_for_tickers ...)>] INFO: Waiting for application startup. INFO: Application startup complete.
这说明lifespan已经执行,定时任务也成功添加了。
额外调试提示(如果还是有问题)
- 确认你的FastAPI版本≥0.95.0:
lifespan是在这个版本才引入的,你用的0.116.1完全符合要求 - 检查
asynccontextmanager的导入:你用的from contextlib import asynccontextmanager是对的,Python 3.7+都支持 - 排查其他启动阻塞:比如
load_dotenv或日志配置是否抛出了静默异常,但从你的日志看Application startup complete,说明这部分没问题
内容来源于stack exchange
相关产品推荐
相关产品推荐

