发现一个 Flask-APScheduler 的一个小坑

我在 bamboo-cms 的模板同步功能中用到了 Flask-APScheduler,在写测试用例的时候遇到了一个报错,如下:

self = <apscheduler.schedulers.background.BackgroundScheduler object at 0x7f8e97a95a10>, gconfig = {}, prefix = 'apscheduler.', options = {}

    def configure(self, gconfig={}, prefix='apscheduler.', **options):
        """
        Reconfigures the scheduler with the given options.
    
        Can only be done when the scheduler isn't running.
    
        :param dict gconfig: a "global" configuration dictionary whose values can be overridden by
            keyword arguments to this method
        :param str|unicode prefix: pick only those keys from ``gconfig`` that are prefixed with
            this string (pass an empty string or ``None`` to use all keys)
        :raises SchedulerAlreadyRunningError: if the scheduler is already running
    
        """
        if self.state != STATE_STOPPED:
>           raise SchedulerAlreadyRunningError
E           apscheduler.schedulers.SchedulerAlreadyRunningError: Scheduler is already running

先上最小可复现例程:

demo/demo/__init__.py

from flask import Flask
from flask_apscheduler import APScheduler

scheduler = APScheduler()


def create_app():
    app = Flask(__name__)
    scheduler.init_app(app)
    scheduler.start()
    return app

demo/tests/conftest.py

from typing import Generator
from flask import Flask

import pytest

from demo import create_app


@pytest.fixture(autouse=True)
def app() -> Generator[Flask, None, None]:
    app = create_app()
    with app.app_context():
        yield app

demo/tests/test_app.py

def test_f1():
    pass


def test_f2():
    pass

我最开始只有一个测试用例,所以不会触发这个坑,但是当我写了两个测试用例之后,我发现每次跑测试用例都只有第一个能跑过,其余的都报 SchedulerAlreadyRunningError,于是猜测和每个测试用例都会创建一个 Flask App 有关,即 demo/tests/conftest.py 定义的 fixture。

我首先怀疑的是 demo/demo/__init__.py 里的第 10 行,即 scheduler.start() 重复执行导致报错,于是我在这一行前面添加检查 APScheduler 是否已经运行的逻辑,但是发现仍然报错,报错信息一致。

然后我开始逐行调试,发现在跑第二个测试用例时,create_app 在跑到第 9 行(scheduler.init_app(app))就报错了,这里算是定位了病因,即 APScheduler 实例已经在运行,此时无法调用 init_app,这个逻辑写在 APScheduler 源代码 apscheduler/schedulers/base.py:117 中。

解决方案:

  1. 把实例化 APScheduler 的逻辑放到 create_app 里,这样每个 Flask App 都有一个专属的 APScheduler,这样似乎会更合理(其他 Flask 扩展是否也应该这样呢? @greyli );
  2. init_app 前判断 APScheduler 是否在运行,在运行就调用 shutdown
  3. 把 demo/tests/conftest.py 下 appscope 改为 session,这个方案可能会不可控,导致测试用例间的数据相互干扰。

目前为了和其他扩展的实例化风格保持一致,我用了方案 2。

应用这个包的目的是?如果是任务队列的话,可以用rq

1 个赞

主要目的是不断去拉模板的更新,试了一下,还是 APScheduler 实现起来方便,rq 得通过递归调用去实现,而且我还遇到了 self 无法被 pickle 序列化的问题。

如果是定时任务,可以flask commands搭配cron使用,所有任务具备app_context,且不需要APScheduler和rq。