使用 APScheduler 计划在 Flask 上每小时运行的函数中的 SQLAlchemy 查询

Question

对于我的烧瓶应用程序，我有一个名为“flaskblog”的文件夹。 在同一个目录中我有 run.py:

from flaskblog import app, db

db.init_app(app)
if __name__ == '__main__':
    app.run(debug=True, port=5000)

在flaskblog文件夹init.py中：

#...some more code
from flask import Flask
app = Flask(__name__)
app.config.from_object(Config)
#.. and so on

def register_app(app):
    """Create an application."""
    main_blueprint = Blueprint('main', __name__)
    app.register_blueprint(main_blueprint)
    return app

register_app(app)

# import BackgroundScheduler
from apscheduler.schedulers.background import BackgroundScheduler
# define the job
from flaskblog.models import Chatmessage
from datetime import datetime, timedelta
def hello_job():
    today = datetime.today()
    start_date = today - timedelta(days=1)
    end_date = today + timedelta(days=1)
    messages = Chatmessage.query.filter(Chatmessage.date_sent <= end_date).\
        filter(Chatmessage.date_sent >= start_date).all()
    
    print(today, start_date, end_date)
    print('Hello Job! The time is: ')
# init BackgroundScheduler job
scheduler = BackgroundScheduler()
# in your case you could change seconds to hours
scheduler.add_job(hello_job, trigger='interval', seconds=10)
scheduler.start()
#find correct place to put job starts

我正在尝试安排一个函数不时运行，就像这个问题中解释的那样： How to schedule a function to run every hour on Flask?

问题是我想使用模型，Flask-SQLAlchemy .. 像 Model.query ......但由于循环导入，在应用程序注册“register_app(app)”之前不能使用它。因此，我在代码中的注释“#import BackgroundScheduler”之后添加了时间表。

这是正确的方法吗？我从重新加载可视代码调试器并看到上次重新加载的函数 hello_job() 仍在运行，或者如果应用程序无法加载函数 hello_job() 无论如何都开始运行，我对此感到怀疑。所以恐怕它可能会在生产服务器上出现问题。

Answer 1

你可以这样做

在你的 init.py 中

...
scheduler = BackgroundScheduler()
scheduler.start()
atexit.register(lambda: scheduler.shutdown())

...
#at the end    
import job

在一个job.py中

from init import scheduler

cron_time = '0 * * * *'

@scheduler.scheduled_job(CronTrigger.from_crontab(cron_time), id='hello_job')
def hello_job():
   ...