Safely create alembic_version table and add the correct stamps

invenio_db/cli.py

@@ -3,6 +3,7 @@
 # This file is part of Invenio.
 # Copyright (C) 2015-2018 CERN.
 # Copyright (C) 2024 Graz University of Technology.
+# Copyright (C) 2026 University of Münster.
 #
 # Invenio is free software; you can redistribute it and/or modify it
 # under the terms of the MIT License; see LICENSE file for more details.
@@ -104,4 +105,5 @@ def destroy():
         except FileNotFoundError:
             click.secho("Sqlite database has not been initialised", fg="red", bold=True)
     else:
+        current_db.engine.dispose()
         drop_database(plain_url)

invenio_db/utils.py

@@ -3,6 +3,7 @@
 # This file is part of Invenio.
 # Copyright (C) 2017-2018 CERN.
 # Copyright (C) 2022-2026 Graz University of Technology.
+# Copyright (C) 2026 University of Münster.
 #
 # Invenio is free software; you can redistribute it and/or modify it
 # under the terms of the MIT License; see LICENSE file for more details.
@@ -13,6 +14,7 @@
 from functools import partial
 
 from alembic import op
+from alembic.migration import MigrationContext
 from flask import current_app
 from sqlalchemy import inspect
 
@@ -67,10 +69,13 @@ def rebuild_encrypted_properties(old_key, model, properties, db=_db):
 def create_alembic_version_table():
     """Create alembic_version table."""
     alembic = current_app.extensions["invenio-db"].alembic
-    if not alembic.migration_context._has_version_table():
-        alembic.migration_context._ensure_version_table()
-        for head in alembic.script_directory.revision_map._real_heads:
-            alembic.migration_context.stamp(alembic.script_directory, head)
+    db = current_app.extensions["sqlalchemy"]
+    with db.engine.begin() as connection:
+        context = MigrationContext.configure(connection)
+        if not context._has_version_table():
+            context._ensure_version_table()
+            all_heads = alembic.script_directory.revision_map._real_heads
+            context.stamp(alembic.script_directory, tuple(all_heads))
 
 
 def drop_alembic_version_table():

tests/test_db.py

@@ -3,7 +3,8 @@
 # This file is part of Invenio.
 # Copyright (C) 2015-2018 CERN.
 # Copyright (C) 2022 RERO.
-# Copyright (C) 2024 Graz University of Technology.
+# Copyright (C) 2024-2026 Graz University of Technology.
+# Copyright (C) 2026 University of Münster.
 #
 # Invenio is free software; you can redistribute it and/or modify it
 # under the terms of the MIT License; see LICENSE file for more details.
@@ -287,10 +288,10 @@ def test_entry_points(db, app):
         assert result.exit_code == 0
 
         result = runner.invoke(db_cmd, ["create", "-v"])
-        assert result.exit_code == 1
+        assert result.exit_code == 0
 
         result = runner.invoke(db_cmd, ["create", "-v"])
-        assert result.exit_code == 1
+        assert result.exit_code == 0
 
         result = runner.invoke(db_cmd, ["drop", "-v", "--yes-i-know"])
         assert result.exit_code == 0
@@ -305,7 +306,7 @@ def test_entry_points(db, app):
         assert result.exit_code == 1
 
         result = runner.invoke(db_cmd, ["drop", "--yes-i-know", "create"])
-        assert result.exit_code == 1
+        assert result.exit_code == 0
 
         result = runner.invoke(db_cmd, ["destroy", "--yes-i-know"])
         assert result.exit_code == 0
@@ -326,10 +327,10 @@ def test_entry_points(db, app):
         assert result.exit_code == 0
 
         result = runner.invoke(db_cmd, ["drop", "-v", "--yes-i-know"])
-        assert result.exit_code == 1
+        assert result.exit_code == 0
 
         result = runner.invoke(db_cmd, ["create", "-v"])
-        assert result.exit_code == 1
+        assert result.exit_code == 0
 
         result = runner.invoke(db_cmd, ["drop", "-v", "--yes-i-know"])
         assert result.exit_code == 0