Контекст приложения Flask для sqlalchemy

Question

Я работаю над небольшим API для отдыха в колбе. Api имеет маршрут, который регистрирует запрос и порождает отдельный поток для запуска в фоновом режиме. Вот код:

def dostuff(scriptname):
    new_thread = threading.Thread(target=executescript,args=(scriptname,))
    new_thread.start()

Поток запускается, но выдает ошибку, когда я пытаюсь вставить в базу данных из функции executetescript. Он жалуется на объект БД, не зарегистрированный в приложении.

Я динамически создаю свое приложение (с 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: 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()

Answer 1

Вы запускаете фоновую задачу в другом потоке, который не имеет контекста приложения. Вы должны передать объект приложения фоновому работнику. Мигель Гринберг приводит пример этого здесь:

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()

В качестве альтернативы (и, вероятно, наилучшего решения) можно было бы фактически установить сеанс SQLAlchemy с локальной областью потока, а не полагаться на контекст запроса Flask-SQLAlchemy.

>>> from sqlalchemy.orm import scoped_session
>>> from sqlalchemy.orm import sessionmaker

>>> session_factory = sessionmaker(bind=some_engine)
>>> Session = scoped_session(session_factory)