This is an automated email from the ASF dual-hosted git repository.
vincbeck pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new e4db0bacb48 Fix fab mysql migration error caused by pre defiend fk
name in orm create (#65831)
e4db0bacb48 is described below
commit e4db0bacb4883d7e008d9fcfa46369ad29990149
Author: Jeongwoo Do <[email protected]>
AuthorDate: Mon Apr 27 23:05:33 2026 +0900
Fix fab mysql migration error caused by pre defiend fk name in orm create
(#65831)
---
.../0001_3_5_0_fix_fab_db_inconsistencies.py | 30 ++++++++---
.../tests/unit/fab/auth_manager/models/test_db.py | 59 ++++++++++++++++++++++
2 files changed, 83 insertions(+), 6 deletions(-)
diff --git
a/providers/fab/src/airflow/providers/fab/migrations/versions/0001_3_5_0_fix_fab_db_inconsistencies.py
b/providers/fab/src/airflow/providers/fab/migrations/versions/0001_3_5_0_fix_fab_db_inconsistencies.py
index e4b843996c1..3913ae1b923 100644
---
a/providers/fab/src/airflow/providers/fab/migrations/versions/0001_3_5_0_fix_fab_db_inconsistencies.py
+++
b/providers/fab/src/airflow/providers/fab/migrations/versions/0001_3_5_0_fix_fab_db_inconsistencies.py
@@ -181,6 +181,15 @@ def _drop_unique_constraint_if_exists(table_name: str,
constraint_name: str) ->
batch_op.drop_constraint(constraint_name, type_="unique")
+def _resolve_fk_name(bind, table_name: str, column_name: str, default: str) ->
str:
+ if bind is None:
+ return default
+ for fk in sa.inspect(bind).get_foreign_keys(table_name):
+ if fk["constrained_columns"] == [column_name]:
+ return fk["name"] or default
+ return default
+
+
def _drop_index_if_exists(table_name: str, index_name: str) -> None:
dialect_name = op.get_context().dialect.name
@@ -247,13 +256,20 @@ def upgrade() -> None:
op.execute(sa.text(_mysql_create_idx_permission_view_id_if_not_exists()))
op.execute(sa.text(_mysql_create_idx_role_id_if_not_exists()))
+ pvr_role_fk = _resolve_fk_name(
+ bind, "ab_permission_view_role", "role_id",
"ab_permission_view_role_role_id_fkey"
+ )
+ pvr_view_fk = _resolve_fk_name(
+ bind,
+ "ab_permission_view_role",
+ "permission_view_id",
+ "ab_permission_view_role_permission_view_id_fkey",
+ )
with op.batch_alter_table(
"ab_permission_view_role", schema=None,
naming_convention=_naming_convention
) as batch_op:
-
batch_op.drop_constraint(batch_op.f("ab_permission_view_role_role_id_fkey"),
type_="foreignkey")
- batch_op.drop_constraint(
- batch_op.f("ab_permission_view_role_permission_view_id_fkey"),
type_="foreignkey"
- )
+ batch_op.drop_constraint(pvr_role_fk, type_="foreignkey")
+ batch_op.drop_constraint(pvr_view_fk, type_="foreignkey")
batch_op.create_foreign_key(
batch_op.f("ab_permission_view_role_permission_view_id_fkey"),
"ab_permission_view",
@@ -285,9 +301,11 @@ def upgrade() -> None:
with op.batch_alter_table("ab_register_user", schema=None) as batch_op:
batch_op.create_unique_constraint(batch_op.f("ab_register_user_email_uq"),
["email"])
+ user_role_fk = _resolve_fk_name(bind, "ab_user_role", "role_id",
"ab_user_role_role_id_fkey")
+ user_user_fk = _resolve_fk_name(bind, "ab_user_role", "user_id",
"ab_user_role_user_id_fkey")
with op.batch_alter_table("ab_user_role", schema=None,
naming_convention=_naming_convention) as batch_op:
- batch_op.drop_constraint(batch_op.f("ab_user_role_role_id_fkey"),
type_="foreignkey")
- batch_op.drop_constraint(batch_op.f("ab_user_role_user_id_fkey"),
type_="foreignkey")
+ batch_op.drop_constraint(user_role_fk, type_="foreignkey")
+ batch_op.drop_constraint(user_user_fk, type_="foreignkey")
batch_op.create_foreign_key(
batch_op.f("ab_user_role_role_id_fkey"), "ab_role", ["role_id"],
["id"], ondelete="CASCADE"
)
diff --git a/providers/fab/tests/unit/fab/auth_manager/models/test_db.py
b/providers/fab/tests/unit/fab/auth_manager/models/test_db.py
index de6bac3defc..36fab389d2d 100644
--- a/providers/fab/tests/unit/fab/auth_manager/models/test_db.py
+++ b/providers/fab/tests/unit/fab/auth_manager/models/test_db.py
@@ -194,5 +194,64 @@ try:
if original_revision and current_revision != original_revision:
manager.upgradedb(to_revision=original_revision)
+ @pytest.mark.backend("mysql")
+ def test_upgradedb_mysql_succeeds_with_implicit_fk_names(self,
session):
+ # Legacy MySQL DBs carry FKs auto-named like ``<table>_ibfk_N``
rather than the
+ # naming-convention names ``batch_op.f()`` produces. The upgrade
must resolve the
+ # live FK names via introspection so the drops issue the correct
identifiers.
+ manager = FABDBManager(session=session)
+ original_revision = manager.get_current_revision()
+
+ try:
+ manager.downgrade(to_revision="6709f7a774b9")
+
+ replacements = {
+ "ab_permission_view_role": [
+ ("permission_view_id", "ab_permission_view", "id"),
+ ("role_id", "ab_role", "id"),
+ ],
+ "ab_user_role": [
+ ("user_id", "ab_user", "id"),
+ ("role_id", "ab_role", "id"),
+ ],
+ }
+ with engine.begin() as conn:
+ for table_name, fks in replacements.items():
+ for fk in
sa.inspect(conn).get_foreign_keys(table_name):
+ if fk["name"]:
+ conn.execute(
+ sa.text(f"ALTER TABLE `{table_name}` DROP
FOREIGN KEY `{fk['name']}`")
+ )
+ for column, ref_table, ref_column in fks:
+ conn.execute(
+ sa.text(
+ f"ALTER TABLE `{table_name}` "
+ f"ADD FOREIGN KEY (`{column}`) "
+ f"REFERENCES `{ref_table}`
(`{ref_column}`) ON DELETE CASCADE"
+ )
+ )
+
+ implicit_fk_names = {
+ fk["name"] for fk in
sa.inspect(engine).get_foreign_keys("ab_permission_view_role")
+ }
+ assert any(name and "_ibfk_" in name for name in
implicit_fk_names)
+
+ manager.upgradedb(to_revision="02ca36b0235b")
+
+ pvr_fk_names = {
+ fk["name"] for fk in
sa.inspect(engine).get_foreign_keys("ab_permission_view_role")
+ }
+ user_role_fk_names = {
+ fk["name"] for fk in
sa.inspect(engine).get_foreign_keys("ab_user_role")
+ }
+ assert "ab_permission_view_role_role_id_fkey" in pvr_fk_names
+ assert "ab_permission_view_role_permission_view_id_fkey" in
pvr_fk_names
+ assert "ab_user_role_role_id_fkey" in user_role_fk_names
+ assert "ab_user_role_user_id_fkey" in user_role_fk_names
+ finally:
+ current_revision = manager.get_current_revision()
+ if original_revision and current_revision != original_revision:
+ manager.upgradedb(to_revision=original_revision)
+
except ModuleNotFoundError:
pass
