Search code examples
pythonflasksqlalchemyflask-sqlalchemyreflect

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


I am having the same issue as this thread: Reflecting tables with Flask-SQLAlchemy raises RuntimeError: application not registered I tried adding:

db.reflect(app=app)

in the db initialization method, but I am still getting:

  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/.

How can I have reflected tables when using AppFactory structure without having those issues?

Models:

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)

app:

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

I spent much time searching for a solution to this issue, but so far I've had no luck. Thanks in advance!


Solution

  • The issue is that the app tries to go through the reflection classes without being fully loaded, so sqlalchemy gets an error due to not finding the currently running app and raises RuntimeError. I found that to fix that you need to provide app_context and therefore this fixed my issue.

    I changed my app file to:

    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