Updates for Flask-SQLAlchemy 3.x

src/flask_migrate/templates/flask-multidb/env.py

@@ -19,6 +19,17 @@
 fileConfig(config.config_file_name)
 logger = logging.getLogger('alembic.env')
 
+
+def get_engine(bind_key=None):
+    try:
+        # this works with Flask-SQLAlchemy>=3
+        return current_app.extensions['migrate'].db.get_engine(
+            bind_key=bind_key)
+    except TypeError:
+        # this works with Flask-SQLAlchemy<3
+        return current_app.extensions['migrate'].db.get_engine(bind=bind_key)
+
+
 # add your model's MetaData object here
 # for 'autogenerate' support
 # from myapp import mymodel
@@ -38,8 +49,7 @@
 for bind in bind_names:
     context.config.set_section_option(
         bind, "sqlalchemy.url",
-        str(current_app.extensions['migrate'].db.get_engine(
-            bind=bind).url).replace('%', '%%'))
+        str(get_engine(bind_key=bind).url).replace('%', '%%'))
 target_db = current_app.extensions['migrate'].db
 
 # other values from the config, defined by the needs of env.py,
@@ -132,8 +142,7 @@ def process_revision_directives(context, revision, directives):
     }
     for name in bind_names:
         engines[name] = rec = {}
-        rec['engine'] = current_app.extensions['migrate'].db.get_engine(
-            bind=name)
+        rec['engine'] = get_engine(bind_key=name)
 
     for name, rec in engines.items():
         engine = rec['engine']

tests/app.py

@@ -1,10 +1,14 @@
 #!/bin/env python
+import os
 from flask import Flask
 from flask_sqlalchemy import SQLAlchemy
 from flask_migrate import Migrate
 
+basedir = os.path.abspath(os.path.dirname(__file__))
+
 app = Flask(__name__)
-app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///app.db'
+app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///' + os.path.join(
+    basedir, 'app.db')
 app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
 
 db = SQLAlchemy(app)

tests/app_compare_type1.py

@@ -1,9 +1,13 @@
+import os
 from flask import Flask
 from flask_sqlalchemy import SQLAlchemy
 from flask_migrate import Migrate
 
+basedir = os.path.abspath(os.path.dirname(__file__))
+
 app = Flask(__name__)
-app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///app.db'
+app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///' + os.path.join(
+    basedir, 'app.db')
 app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
 
 db = SQLAlchemy(app)

tests/app_compare_type2.py

@@ -1,9 +1,13 @@
+import os
 from flask import Flask
 from flask_sqlalchemy import SQLAlchemy
 from flask_migrate import Migrate
 
+basedir = os.path.abspath(os.path.dirname(__file__))
+
 app = Flask(__name__)
-app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///app.db'
+app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///' + os.path.join(
+    basedir, 'app.db')
 app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
 
 db = SQLAlchemy(app)

tests/app_custom_directory.py

@@ -1,9 +1,13 @@
+import os
 from flask import Flask
 from flask_sqlalchemy import SQLAlchemy
 from flask_migrate import Migrate
 
+basedir = os.path.abspath(os.path.dirname(__file__))
+
 app = Flask(__name__)
-app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///app.db'
+app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///' + os.path.join(
+    basedir, 'app.db')
 app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
 
 db = SQLAlchemy(app)

tests/app_custom_directory_path.py

@@ -1,10 +1,14 @@
+import os
 from flask import Flask
 from flask_sqlalchemy import SQLAlchemy
 from flask_migrate import Migrate
 from pathlib import Path
 
+basedir = os.path.abspath(os.path.dirname(__file__))
+
 app = Flask(__name__)
-app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///app.db'
+app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///' + os.path.join(
+    basedir, 'app.db')
 app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
 
 db = SQLAlchemy(app)

tests/app_multidb.py

@@ -1,13 +1,17 @@
 #!/bin/env python
+import os
 from flask import Flask
 from flask_sqlalchemy import SQLAlchemy
 from flask_migrate import Migrate
 
+basedir = os.path.abspath(os.path.dirname(__file__))
+
 app = Flask(__name__)
-app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///app1.db'
+app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///' + os.path.join(
+    basedir, 'app1.db')
 app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
 app.config['SQLALCHEMY_BINDS'] = {
-    "db1": "sqlite:///app2.db",
+    "db1": "sqlite:///" + os.path.join(basedir, "app2.db"),
 }
 
 db = SQLAlchemy(app)

tests/test_custom_template.py

@@ -61,9 +61,10 @@ def test_migrate_upgrade(self):
         (o, e, s) = run_cmd('app.py', 'flask db upgrade')
         self.assertTrue(s == 0)
 
-        from .app import db, User
-        db.session.add(User(name='test'))
-        db.session.commit()
+        from .app import app, db, User
+        with app.app_context():
+            db.session.add(User(name='test'))
+            db.session.commit()
 
         with open('migrations/README', 'rt') as f:
             assert f.readline().strip() == 'Custom template.'

tests/test_migrate.py

@@ -61,9 +61,10 @@ def test_migrate_upgrade(self):
         (o, e, s) = run_cmd('app.py', 'flask db upgrade')
         self.assertTrue(s == 0)
 
-        from .app import db, User
-        db.session.add(User(name='test'))
-        db.session.commit()
+        from .app import app, db, User
+        with app.app_context():
+            db.session.add(User(name='test'))
+            db.session.commit()
 
     def test_custom_directory(self):
         (o, e, s) = run_cmd('app_custom_directory.py', 'flask db init')
@@ -73,9 +74,10 @@ def test_custom_directory(self):
         (o, e, s) = run_cmd('app_custom_directory.py', 'flask db upgrade')
         self.assertTrue(s == 0)
 
-        from .app_custom_directory import db, User
-        db.session.add(User(name='test'))
-        db.session.commit()
+        from .app_custom_directory import app, db, User
+        with app.app_context():
+            db.session.add(User(name='test'))
+            db.session.commit()
 
     def test_custom_directory_path(self):
         (o, e, s) = run_cmd('app_custom_directory_path.py', 'flask db init')
@@ -85,9 +87,10 @@ def test_custom_directory_path(self):
         (o, e, s) = run_cmd('app_custom_directory_path.py', 'flask db upgrade')
         self.assertTrue(s == 0)
 
-        from .app_custom_directory_path import db, User
-        db.session.add(User(name='test'))
-        db.session.commit()
+        from .app_custom_directory_path import app, db, User
+        with app.app_context():
+            db.session.add(User(name='test'))
+            db.session.commit()
 
     def test_compare_type(self):
         (o, e, s) = run_cmd('app_compare_type1.py', 'flask db init')

tests/test_multidb_migrate.py

@@ -12,6 +12,9 @@ def run_cmd(app, cmd):
     process = subprocess.Popen(shlex.split(cmd), stdout=subprocess.PIPE,
                                stderr=subprocess.PIPE)
     (stdout, stderr) = process.communicate()
+    print('\n$ ' + cmd)
+    print(stdout.decode('utf-8'))
+    print(stderr.decode('utf-8'))
     return stdout, stderr, process.wait()
 
 
@@ -65,10 +68,11 @@ def test_multidb_migrate_upgrade(self):
         self.assertIn(('group',), tables)
 
         # ensure the databases can be written to
-        from .app_multidb import db, User, Group
-        db.session.add(User(name='test'))
-        db.session.add(Group(name='group'))
-        db.session.commit()
+        from .app_multidb import app, db, User, Group
+        with app.app_context():
+            db.session.add(User(name='test'))
+            db.session.add(Group(name='group'))
+            db.session.commit()
 
         # ensure the downgrade works
         (o, e, s) = run_cmd('app_multidb.py', 'flask db downgrade')