python – sqlalchemy的Flask应用程序上下文
发布时间:2020-12-16 23:09:43 所属栏目:Python 来源:网络整理
导读:我正在研究烧瓶中的小休息api. Api具有注册请求并生成单独线程以在后台运行的路由.这是代码: def dostuff(scriptname): new_thread = threading.Thread(target=executescript,args=(scriptname,)) new_thread.start() 线程启动但是当我尝试从executioncript
我正在研究烧瓶中的小休息api. Api具有注册请求并生成单独线程以在后台运行的路由.这是代码:
def dostuff(scriptname): new_thread = threading.Thread(target=executescript,args=(scriptname,)) new_thread.start() 线程启动但是当我尝试从executioncript函数插入db时它会出错.它抱怨db对象没有在应用程序中注册. 我正在动态创建我的应用程序(使用api作为Blueprint). 这是应用程序的结构 -run.py ## runner script -config -development.py -prod.py -app -__init__.py - auth.py - api_v1 - __init__.py - routes.py - models.py 这是我的跑步者脚本run.py: from app import create_app,db if __name__ == '__main__': app = create_app(os.environ.get('FLASK_CONFIG','development')) with app.app_context(): db.create_all() app.run() 以下是来自app / __ init__.py的代码,用于创建应用: from flask import Flask,jsonify,g from flask.ext.sqlalchemy import SQLAlchemy db = SQLAlchemy() def create_app(config_name): """Create an application instance.""" app = Flask(__name__) # apply configuration cfg = os.path.join(os.getcwd(),'config',config_name + '.py') app.config.from_pyfile(cfg) # initialize extensions db.init_app(app) # register blueprints from .api_v1 import api as api_blueprint app.register_blueprint(api_blueprint,url_prefix='/api/') return app 我需要知道的是如何在routes.py中扩展应用程序上下文.我不能直接在那里导入应用程序,如果我执行以下操作,我会得到RuntimeError:在应用程序上下文之外工作 def executescript(scriptname): with current_app.app_context(): test_run = Testrun(pid=989,exit_status=988,comments="Test TestStarted") db.session.add(test_run) db.session.commit() 解决方法
您正在另一个没有应用程序上下文的线程中运行后台任务.您应该将app对象传递给后台worker.
Miguel Grinberg gives an example of this here:
from threading import Thread from app import app def send_async_email(app,msg): with app.app_context(): mail.send(msg) def send_email(subject,sender,recipients,text_body,html_body): msg = Message(subject,sender=sender,recipients=recipients) msg.body = text_body msg.html = html_body thr = Thread(target=send_async_email,args=[app,msg]) thr.start() 或者(可能是最好的解决方案)实际上是set up a thread-local scoped SQLAlchemy session,而不是依赖于Flask-SQLAlchemy的请求上下文. >>> from sqlalchemy.orm import scoped_session >>> from sqlalchemy.orm import sessionmaker >>> session_factory = sessionmaker(bind=some_engine) >>> Session = scoped_session(session_factory) (编辑:李大同) 【声明】本站内容均来自网络,其相关言论仅代表作者个人观点,不代表本站立场。若无意侵犯到您的权利,请及时与联系站长删除相关内容! |