sqlalchemy 的 Flask 应用上下文
Flask app context for sqlalchemy
我正在 Flask 中做一个小休息 api。 Api 具有注册请求并在后台生成单独线程到 运行 的路由。这是代码:
def dostuff(scriptname):
new_thread = threading.Thread(target=executescript,args=(scriptname,))
new_thread.start()
线程已启动,但当我尝试从 executescript 函数插入数据库时出错。它抱怨未在应用程序中注册数据库对象。
我正在动态创建我的应用程序(使用 api 作为蓝图)。
这是应用程序的结构
-run.py ## runner script
-config
-development.py
-prod.py
-app
-__init__.py
- auth.py
- api_v1
- __init__.py
- routes.py
- models.py
这是我的 运行ner 脚本 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: working outside of application context
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 对象传递给后台工作人员。 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)
我正在 Flask 中做一个小休息 api。 Api 具有注册请求并在后台生成单独线程到 运行 的路由。这是代码:
def dostuff(scriptname):
new_thread = threading.Thread(target=executescript,args=(scriptname,))
new_thread.start()
线程已启动,但当我尝试从 executescript 函数插入数据库时出错。它抱怨未在应用程序中注册数据库对象。
我正在动态创建我的应用程序(使用 api 作为蓝图)。
这是应用程序的结构
-run.py ## runner script
-config
-development.py
-prod.py
-app
-__init__.py
- auth.py
- api_v1
- __init__.py
- routes.py
- models.py
这是我的 运行ner 脚本 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: working outside of application context
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 对象传递给后台工作人员。 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)