我正在研究烧瓶中的小休息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)