This is an automated email from the ASF dual-hosted git repository. kaxilnaik pushed a commit to branch v3-0-test in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 60df6ddafa273fc417013a1a0cfa734e3958e431 Author: Ash Berlin-Taylor <[email protected]> AuthorDate: Mon Aug 18 13:47:03 2025 +0100 [v3-0-test] Support directly calling `mask_secret` from with in Triggers (#54612) Somewhat annoyingly it has to be `sync_to_async(mask_secret)` but that is unavoidable unfortunately. Similar to #54574, but for the trigger. (cherry picked from commit a2bd6255e9bea6a82180f848b8ebedc58b2ab976) --- airflow-core/src/airflow/jobs/triggerer_job_runner.py | 6 ++++++ airflow-core/tests/unit/jobs/test_triggerer_job.py | 2 ++ 2 files changed, 8 insertions(+) diff --git a/airflow-core/src/airflow/jobs/triggerer_job_runner.py b/airflow-core/src/airflow/jobs/triggerer_job_runner.py index 916854fdf42..ddcee486680 100644 --- a/airflow-core/src/airflow/jobs/triggerer_job_runner.py +++ b/airflow-core/src/airflow/jobs/triggerer_job_runner.py @@ -58,6 +58,7 @@ from airflow.sdk.execution_time.comms import ( GetTICount, GetVariable, GetXCom, + MaskSecret, OKResponse, PutVariable, SetXCom, @@ -250,6 +251,7 @@ ToTriggerSupervisor = Annotated[ GetTaskStates, GetDagRunState, GetDRCount, + MaskSecret, ], Field(discriminator="type"), ] @@ -472,6 +474,10 @@ class TriggerRunnerSupervisor(WatchedSubprocess): resp = TaskStatesResult.from_api_response(run_id_task_state_map) else: resp = run_id_task_state_map + elif isinstance(msg, MaskSecret): + from airflow.sdk.execution_time.secrets_masker import mask_secret + + mask_secret(msg.value, msg.name) else: raise ValueError(f"Unknown message type {type(msg)}") diff --git a/airflow-core/tests/unit/jobs/test_triggerer_job.py b/airflow-core/tests/unit/jobs/test_triggerer_job.py index 89f5823f658..4265d6e2c40 100644 --- a/airflow-core/tests/unit/jobs/test_triggerer_job.py +++ b/airflow-core/tests/unit/jobs/test_triggerer_job.py @@ -626,11 +626,13 @@ class CustomTrigger(BaseTrigger): from airflow.sdk import Variable from airflow.sdk.execution_time.xcom import XCom + from airflow.sdk.log import mask_secret conn = await sync_to_async(BaseHook.get_connection)("test_connection") self.log.info("Loaded conn %s", conn.conn_id) get_variable_value = await sync_to_async(Variable.get)("test_get_variable") + await sync_to_async(mask_secret)(get_variable_value) self.log.info("Loaded variable %s", get_variable_value) get_xcom_value = await sync_to_async(XCom.get_one)(
