FastAPI中使用AsyncIOScheduler与SQLAlchemyJobStore启动报错求助

问题说明 #

在FastAPI异步环境中使用APScheduler的AsyncIOScheduler和SQLAlchemyJobStore时，调用Schedule.start()触发sqlalchemy.exc.MissingGreenlet异常。尝试过给start()添加async/await、更换MySQL数据库URL，问题仍未解决。

错误栈 #

File "D:\app\endpoints\admin\scheduler.py", line 23, in <module>
    Schedule.start()
  File "D:\intergration-server\venv\Lib\site-packages\apscheduler\schedulers\asyncio.py", line 37, in start
    super(AsyncIOScheduler, self).start(paused)
  File "D:\intergration-server\venv\Lib\site-packages\apscheduler\schedulers\base.py", line 173, in start
    store.start(self, alias)
  File "D:\intergration-server\venv\Lib\site-packages\apscheduler\jobstores\sqlalchemy.py", line 68, in start
    self.jobs_t.create(self.engine, True)
  File "D:\intergration-server\venv\Lib\site-packages\sqlalchemy\sql\schema.py", line 1293, in create
    bind._run_ddl_visitor(ddl.SchemaGenerator, self, checkfirst=checkfirst)
  File "D:\intergration-server\venv\Lib\site-packages\sqlalchemy\engine\base.py", line 3242, in _run_ddl_visitor
    with self.begin() as conn:
  File "C:\Program Files\WindowsApps\PythonSoftwareFoundation.Python.3.11_3.11.2032.0_x64__qbz5n2kfra8p0\Lib\contextlib.py", line 137, in __enter__
    return next(self.gen)
           ^^^^^^^^^^^^^^
  File "D:\intergration-server\venv\Lib\site-packages\sqlalchemy\engine\base.py", line 3232, in begin
    with self.connect() as conn:
         ^^^^^^^^^^^^^^
  File "D:\intergration-server\venv\Lib\site-packages\sqlalchemy\engine\base.py", line 3268, in connect
    return self._connection_cls(self)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "D:\intergration-server\venv\Lib\site-packages\sqlalchemy\engine\base.py", line 145, in __init__
    self._dbapi_connection = engine.raw_connection()
                             ^^^^^^^^^^^^^^^^^^^^^^^
  File "D:\intergration-server\venv\Lib\site-packages\sqlalchemy\engine\base.py", line 3292, in raw_connection
    return self.pool.connect()
           ^^^^^^^^^^^^^^^^^^^
  File "D:\intergration-server\venv\Lib\site-packages\sqlalchemy\pool\base.py", line 452, in connect
    return _ConnectionFairy._checkout(self)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "D:\intergration-server\venv\Lib\site-packages\sqlalchemy\pool\base.py", line 1269, in _checkout
    fairy = _ConnectionRecord.checkout(pool)
            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "D:\intergration-server\venv\Lib\site-packages\sqlalchemy\pool\base.py", line 716, in checkout
    rec = pool._do_get()
          ^^^^^^^^^^^^^^
  File "D:\intergration-server\venv\Lib\site-packages\sqlalchemy\pool\impl.py", line 169, in _do_get
    with util.safe_reraise():
  File "D:\intergration-server\venv\Lib\site-packages\sqlalchemy\util\langhelpers.py", line 146, in __exit__
    raise exc_value.with_traceback(exc_tb)
  File "D:\intergration-server\venv\Lib\site-packages\sqlalchemy\pool\impl.py", line 167, in _do_get
    return self._create_connection()
           ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "D:\intergration-server\venv\Lib\site-packages\sqlalchemy\pool\base.py", line 393, in _create_connection
    return _ConnectionRecord(self)
           ^^^^^^^^^^^^^^^^^^^^^^^
  File "D:\intergration-server\venv\Lib\site-packages\sqlalchemy\pool\base.py", line 678, in __init__
    self.__connect()
  File "D:\intergration-server\venv\Lib\site-packages\sqlalchemy\pool\base.py", line 902, in __connect
    with util.safe_reraise():
  File "D:\intergration-server\venv\Lib\site-packages\sqlalchemy\util\langhelpers.py", line 146, in __exit__
    raise exc_value.with_traceback(exc_tb)
  File "D:\intergration-server\venv\Lib\site-packages\sqlalchemy\pool\base.py", line 898, in __connect
    self.dbapi_connection = connection = pool._invoke_creator(self)
                                         ^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "D:\intergration-server\venv\Lib\site-packages\sqlalchemy\engine\create.py", line 637, in connect
    return dialect.connect(*cargs, **cparams)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "D:\intergration-server\venv\Lib\site-packages\sqlalchemy\engine\default.py", line 616, in connect
    return self.loaded_dbapi.connect(*cargs, **cparams)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "D:\intergration-server\venv\Lib\site-packages\sqlalchemy\dialects\mysql\asyncmy.py", line 281, in connect
    await_only(creator_fn(*arg, **kw)),
    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "D:\intergration-server\venv\Lib\site-packages\sqlalchemy\util\_concurrency_py3k.py", line 116, in await_only
    raise exc.MissingGreenlet(
sqlalchemy.exc.MissingGreenlet: greenlet_spawn has not been called; can't call await_only() here. Was IO attempted in an unexpected place? (Background on this error at: https://sqlalche.me/e/20/xd2s)

相关代码 #

jobstore = {
    "default": SQLAlchemyJobStore(url=manager.SQLALCHEMY_DATABASE_URL, engine_options={"pool_pre_ping": True}),
}

Schedule = AsyncIOScheduler(jobstores=jobstore, timezone="UTC")
Schedule.start()


def resp_ok(*, code=0, msg="ok", data: Union[list, dict, str] = None) -> dict:
    return {"code": code, "msg": msg, "data": data}


def resp_fail(*, code=1, msg="fail", data: Union[list, dict, str] = None):
    return {"code": code, "msg": msg, "data": data}


def cron_task(a1: str) -> None:
    print(a1, time.strftime("'%Y-%m-%d %H:%M:%S'"))


@router.get("/jobs/all", tags=["schedule"], summary="获取所有job信息")
async def get_scheduled_syncs():
    schedules = []
    for job in Schedule.get_jobs():
        schedules.append(
            {
                "job_id": job.id,
                "func_name": job.func_ref,
                "func_args": job.args,
                "cron_model": str(job.trigger),
                "next_run": str(job.next_run_time),
            }
        )
    return resp_ok(data=schedules)

解决方案 #

核心原因 #

SQLAlchemyJobStore默认使用同步SQLAlchemy引擎，而AsyncIOScheduler启动时在主线程执行数据库DDL操作，此时FastAPI的异步绿程上下文尚未初始化，导致SQLAlchemy异步驱动无法找到绿程环境，抛出MissingGreenlet异常。

方案一：使用异步版JobStore（推荐） #

APScheduler 3.10+版本提供了AsyncSQLAlchemyJobStore，专门适配异步环境。修改jobstore配置：

from apscheduler.jobstores.async_sqlalchemy import AsyncSQLAlchemyJobStore

jobstore = {
    "default": AsyncSQLAlchemyJobStore(url=manager.SQLALCHEMY_DATABASE_URL, engine_options={"pool_pre_ping": True}),
}

Schedule = AsyncIOScheduler(jobstores=jobstore, timezone="UTC")

注意确保数据库URL使用异步驱动格式（如mysql+asyncmy://user:password@host/dbname）。

方案二：在FastAPI启动事件中启动调度器 #

将调度器启动逻辑放到FastAPI的startup异步事件中，确保在绿程上下文环境中执行：

from fastapi import FastAPI
import asyncio

app = FastAPI()

jobstore = {
    "default": SQLAlchemyJobStore(url=manager.SQLALCHEMY_DATABASE_URL, engine_options={"pool_pre_ping": True}),
}
Schedule = AsyncIOScheduler(jobstores=jobstore, timezone="UTC")

@app.on_event("startup")
async def startup_event():
    # 使用call_soon_threadsafe避免阻塞异步事件循环
    loop = asyncio.get_event_loop()
    loop.call_soon_threadsafe(Schedule.start)

# 保留原有路由代码...

额外注意事项 #

• 不要在模块加载阶段直接调用Schedule.start()，此时异步上下文未初始化，必然触发绿程异常
• 确保安装了对应异步数据库驱动（如asyncmy用于MySQL）
• 若使用旧版APScheduler无法升级，可考虑改用同步调度器（如BackgroundScheduler）并在单独线程运行，但不推荐在异步FastAPI中混用同步线程

内容的提问来源于stack exchange，提问作者Sarim Sikander