(airflow) 04/05: Fix decryption of trigger kwargs when downgrading. (#38743)

[ep...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

ephraimanierobi pushed a commit to branch v2-9-test
in repository https://gitbox.apache.org/repos/asf/airflow.git

commit cbe61bc0cc028efe2657db1841b9371230eaa7e5
Author: Ephraim Anierobi <sp...@gmail.com>
AuthorDate: Thu Apr 4 18:24:29 2024 +0100

    Fix decryption of trigger kwargs when downgrading. (#38743)
    
    * Fix decryption of trigger kwargs when downgrading.
    
    This failed because the query for Triggers after the downgrade lazy
    loads the taskinstance table(ORM) which doesn't have the task_display_name
    column at that downloaded point.
    
    The fix was to query specifically on the encrypted_kwargs column.
    
    * Properly positon the decryption after downgrade and not in offline migration
    
    (cherry picked from commit 567246f1b1111be5a9d8a85ec3be7a103341101d)
---
 airflow/utils/db.py | 14 +++++++-------
 1 file changed, 7 insertions(+), 7 deletions(-)

diff --git a/airflow/utils/db.py b/airflow/utils/db.py
index 26697bb118..c0d282a587 100644
--- a/airflow/utils/db.py
+++ b/airflow/utils/db.py
@@ -993,7 +993,7 @@ def decrypt_trigger_kwargs(*, session: Session) -> None:
         # this can happen when we downgrade to an old version before the Trigger table was added
         return
 
-    for trigger in session.query(Trigger):
+    for trigger in session.scalars(select(Trigger.encrypted_kwargs)):
         # decrypt the string and convert it to serialized dict
         trigger.encrypted_kwargs = json.dumps(BaseSerialization.serialize(trigger.kwargs))
     session.commit()
@@ -1744,12 +1744,12 @@ def downgrade(*, to_revision, from_revision=None, show_sql_only=False, session:
         else:
             log.info("Applying downgrade migrations.")
             command.downgrade(config, revision=to_revision, sql=show_sql_only)
-    if _revision_greater(
-        config,
-        _REVISION_HEADS_MAP["2.9.0"],
-        to_revision,
-    ):
-        decrypt_trigger_kwargs(session=session)
+            if _revision_greater(
+                config,
+                _REVISION_HEADS_MAP["2.9.0"],
+                to_revision,
+            ):
+                decrypt_trigger_kwargs(session=session)
 
 
 def drop_airflow_models(connection):