python-3.x - 如何使用 APScheduler 在 Flask 中修复“弹出错误的应用程序上下文”
问题描述
我根据 API 调用在运行时使用 APScheduler 添加后台任务。换句话说,应用程序启动时没有后台任务。当用户调用 API 时,会在运行时添加任务。但我收到一条错误消息:
AssertionError:弹出错误的应用程序上下文
如果我注释掉安排后台任务的行,该应用程序就可以正常工作。我的应用程序结构如下:
/project
manage.py
requirements.txt
/app
/models
/routes
/utils
/api
config.py
__init__.py
我的manage.py
文件如下所示:
app = create_app('dev')
app.app_context().push()
manager = Manager(app)
migrate = Migrate(app, db, render_as_batch=True)
manager.add_command('db', MigrateCommand)
with app.app_context():
scheduler = BackgroundScheduler()
scheduler.start()
@manager.command
def run():
app.run()
atexit.register(lambda: scheduler.shutdown())
if __name__ == '__main__':
manager.run()
app文件夹内的init .py是:
from flask import Flask
from flask_restful import Api
from flask_sqlalchemy import SQLAlchemy
from email_scheduler.routes.routes import set_routes
from .config import config_by_name
# from app.models.task import TaskModel
db = SQLAlchemy()
def create_app(config_name):
app = Flask(__name__)
app.config.from_object(config_by_name[config_name])
api = Api(app)
set_routes(api)
from email_scheduler.models.api_models import TaskModel, User
db.init_app(app)
with app.app_context():
db.create_all()
return app
我的 api.py 文件是:
class SignUp(Resource):
def clean_scheduling_time(self, schedule_time):
day = schedule_time.split(' ')[0].lower()[:3]
hour, mins = schedule_time.split(' ')[1].split(':')
return day, hour, mins
def post(self):
args = user_parser.parse_args()
username, password = args.get('username'), args.get('password')
schedule_time, email_to = args.get('schedule_time'), args.get('email_to')
if username is None or password is None:
abort(400) # missing arguments
from email_scheduler.models.api_models import User
if User.query.filter_by(username=username).first() is not None:
abort(400) # existing user
user = User(username=username, schedule_time=schedule_time.split(' ')[1], email_to=email_to)
user.hash_password(password)
user.save_to_db()
from manage import scheduler
from email_scheduler.utils.utils import send_email
day, hour, mins = self.clean_scheduling_time(args.get('schedule_time'))
trigger = CronTrigger(day_of_week=day, hour=int(hour), minute=int(mins))
scheduler.add_job(send_email, trigger=trigger)
print(scheduler.get_jobs())
return make_response(jsonify({'username': username}), 200)
奇怪的是,即使我在终端上收到此错误,任务还是会以某种方式被安排并运行。如果我从 api 中取出调度任务的代码,API 运行得很好。我究竟做错了什么?
解决方案
问题出在您的manage.py
文件中。
您正在全局运行以下行:
app.app_context().push()
您正确需要工作人员才能访问应用程序上下文。将其移动到工作人员调用的函数中。
即不是这个:
app = create_app()
app.app_context().push()
def your_async_fn():
# your code for the worker...
但是这个:
def your_async_fn():
app = create_app()
app.app_context().push()
# your code for the worker...
推荐阅读
- reactjs - 无法使用 UseAxios 构建 Gatsby,它引发错误:UnhandledPromiseRejectionWarning:错误:连接 ECONNREFUSED 127.0.0.1:80
- javascript - Webpack 无法在 Electron 应用程序中从 Nodejs 打包 https
- python - 不支持将字符串转换为浮点数 - 时间序列数据的去噪自动编码器
- c - 如何使用 CMake (Mac M1) 导入 Homebrew SDL2_image
- go - 检查 interface{} 是否指向结构
- c - 正确地将浮点数输入到结构中并打印其他提示(scanf 和 fgets 之间的区别)?
- filesystems - 仅当没有相同的文件/目录时才提取 tar
- python - nltk标签的问题
- amazon-web-services - 服务无法承担角色错误 - 使用 JDBC 目标创建 AWS Glue cloudformation
- nasm - 为什么除法后的输出是“@”符号?