使用 AppFactory Structure 时使用 Flask-Sqlalchemy 反映表会引发运行时错误

Reflecting tables with Flask-Sqlalchemy when using AppFactory Structure raises Runtime errors

我遇到了与此线程相同的问题: 我尝试添加:

db.reflect(app=app)

在数据库初始化方法中,但我仍然得到:

  File "C:\Users\myuser\PycharmProjects\server\app\models\__init__.py", line 11, in Entry
    __table__ = db.Table('entry', db.metadata, autoload=True, autoload_with=db.engine)
  File "C:\Users\myuser\PycharmProjects\server\venv\lib\site-packages\flask_sqlalchemy\__init__.py", line 998, in engine
    return self.get_engine()
  File "C:\Users\myuser\PycharmProjects\server\venv\lib\site-packages\flask_sqlalchemy\__init__.py", line 1007, in get_engine
    app = self.get_app(app)
  File "C:\Users\myuser\PycharmProjects\server\venv\lib\site-packages\flask_sqlalchemy\__init__.py", line 1042, in get_app
    raise RuntimeError(
RuntimeError: No application found. Either work inside a view function or push an application context. See http://flask-sqlalchemy.pocoo.org/contexts/.

如何在使用 AppFactory 结构时反射表而不出现这些问题?

型号:

from app.extensions import db

class Entry(db.Model):
    __table__ = db.Table('entry', db.metadata, autoload=True, autoload_with=db.engine)


class Room(db.Model):
    __table__ = db.Table('room', db.metadata, autoload=True, autoload_with=db.engine)

应用程序:

from os import getenv

from flask import Flask, make_response, jsonify
from flask_cors import CORS

from app.extensions import db, migrate, socketio

def create_app(blueprints=None) -> Flask:
    """
    Builds up a Flask app and return it to the caller
    :param blueprints: a custom list of Flask blueprints
    :return: Flask app object
    """
    if blueprints is None:
        blueprints = DEFAULT_BLUEPRINTS

    app = Flask(__name__)

    configure_app(app)
    configure_extensions(app)
    configure_error_handlers(app)
    configure_blueprints(app, blueprints)

    print("Returning app...")

    return app


def configure_app(app):
    app.config.from_object(getenv('APP_SETTINGS', "config.DevelopmentConfig"))
    app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False


def configure_blueprints(app, blueprints):
    for blueprint in blueprints:
        app.register_blueprint(blueprint)


def configure_extensions(app):
    db.init_app(app)

    migrate.init_app(app, db, directory="migrations")

    socketio.init_app(app, cors_allowed_origins='*')

    CORS(app, resources={r"*": {"origins": "*"}})


def configure_error_handlers(app):
    @app.errorhandler(404)
    def not_found(error):
        return make_response(jsonify({'error': 'Not found'}), 404)

    # Return validation errors as JSON
    @app.errorhandler(422)
    @app.errorhandler(400)
    def handle_error(err):
        headers = err.data.get("headers", None)
        messages = err.data.get("messages", ["Invalid request."])
        if headers:
            return jsonify({"errors": messages}), err.code, headers
        else:
            return jsonify({"errors": messages}), err.code

我花了很多时间寻找这个问题的解决方案,但到目前为止我没有运气。提前致谢!

问题是应用程序试图在未完全加载的情况下通过反射 类,因此 sqlalchemy 因未找到当前 运行 应用程序而出现错误并引发 RuntimeError。我发现要解决这个问题,您需要提供 app_context,因此这解决了我的问题。

我将我的应用程序文件更改为:

from os import getenv

from flask import Flask, make_response, jsonify
from flask_cors import CORS

from app.extensions import db, migrate, socketio

def create_app(blueprints=None) -> Flask:
    """
    Builds up a Flask app and return it to the caller
    :param blueprints: a custom list of Flask blueprints
    :return: Flask app object
    """
    if blueprints is None:
        blueprints = DEFAULT_BLUEPRINTS

    app = Flask(__name__)

    configure_app(app)
    configure_extensions(app)
    configure_error_handlers(app)

    with app.app_context():
        # Import blueprints here
        DEFAULT_BLUEPRINTS = [] # list of the blueprints here 

        if blueprints is None:
            blueprints = DEFAULT_BLUEPRINTS

        configure_blueprints(app, blueprints)

    print("Returning app...")

    return app


def configure_app(app):
    app.config.from_object(getenv('APP_SETTINGS', "config.DevelopmentConfig"))
    app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False


def configure_blueprints(app, blueprints):
    for blueprint in blueprints:
        app.register_blueprint(blueprint)


def configure_extensions(app):
    db.init_app(app)

    migrate.init_app(app, db, directory="migrations")

    socketio.init_app(app, cors_allowed_origins='*')

    CORS(app, resources={r"*": {"origins": "*"}})


def configure_error_handlers(app):
    @app.errorhandler(404)
    def not_found(error):
        return make_response(jsonify({'error': 'Not found'}), 404)

    # Return validation errors as JSON
    @app.errorhandler(422)
    @app.errorhandler(400)
    def handle_error(err):
        headers = err.data.get("headers", None)
        messages = err.data.get("messages", ["Invalid request."])
        if headers:
            return jsonify({"errors": messages}), err.code, headers
        else:
            return jsonify({"errors": messages}), err.code