You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by ka...@apache.org on 2025/10/31 14:41:26 UTC
(airflow) branch v3-1-test updated (73336aa0eb0 -> faeb1bf3c20)
This is an automated email from the ASF dual-hosted git repository.
kaxilnaik pushed a change to branch v3-1-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
from 73336aa0eb0 [v3-1-test] SerDe: Check more strictly for pydantic model (#56758) (#57616)
new 860aa6f8ab8 Reduce log level for Connection not found event from error to debug in task-sdk. (#57548)
new faeb1bf3c20 Fix tasks unable to access triggering_user_name` (#56193)
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../execution_api/datamodels/taskinstance.py | 1 +
.../api_fastapi/execution_api/versions/__init__.py | 4 +-
.../versions/{v2025_10_27.py => v2025_11_05.py} | 17 +++---
.../versions/head/test_task_instances.py | 57 +++++++++++++++++++
.../versions/v2025_09_23/test_task_instances.py | 65 ++++++++++++++++++++++
task-sdk/src/airflow/sdk/api/client.py | 2 +-
.../src/airflow/sdk/api/datamodels/_generated.py | 3 +-
task-sdk/src/airflow/sdk/types.py | 1 +
.../task_sdk/execution_time/test_supervisor.py | 2 +
9 files changed, 140 insertions(+), 12 deletions(-)
copy airflow-core/src/airflow/api_fastapi/execution_api/versions/{v2025_10_27.py => v2025_11_05.py} (66%)
(airflow) 02/02: Fix tasks unable to access triggering_user_name` (#56193)
Posted by ka...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
kaxilnaik pushed a commit to branch v3-1-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit faeb1bf3c204de4d69db51ae1cf6e7cc8b34148c
Author: Dheeraj Turaga <dh...@gmail.com>
AuthorDate: Mon Oct 13 18:44:57 2025 -0500
Fix tasks unable to access triggering_user_name` (#56193)
* Add triggering_user_name to DagRunProtocol interface
Enable tasks running in isolated environments to access the username
that triggered a DAG run without requiring direct database access.
Changes:
- Add triggering_user_name field to DagRunProtocol interface
- Update task-sdk DagRun data model to include triggering_user_name
- Update execution API DagRun model to include triggering_user_name
- Ensure field flows from database through API to task context
This allows developers to identify DAG run triggers from within task
code using the existing DagRunProtocol interface.
* Fix tests!
* Add Cadwyn migration for triggering_user_name field in DagRunProtocol
Add backward compatibility migration for the new triggering_user_name field
added to the DagRun model in the Execution API. This ensures older API
clients continue to work seamlessly with newer API servers.
* Add tests for older versions where triggering user is missing
* Remove triggering user from v2025-04-28
* Add a test with triggering_user_name set
(cherry picked from commit 3120146f0a2720d186513282f9e8d807f635434f)
---
.../execution_api/datamodels/taskinstance.py | 1 +
.../api_fastapi/execution_api/versions/__init__.py | 4 +-
.../execution_api/versions/v2025_11_05.py | 36 ++++++++++++
.../versions/head/test_task_instances.py | 57 +++++++++++++++++++
.../versions/v2025_09_23/test_task_instances.py | 65 ++++++++++++++++++++++
.../src/airflow/sdk/api/datamodels/_generated.py | 3 +-
task-sdk/src/airflow/sdk/types.py | 1 +
.../task_sdk/execution_time/test_supervisor.py | 2 +
8 files changed, 167 insertions(+), 2 deletions(-)
diff --git a/airflow-core/src/airflow/api_fastapi/execution_api/datamodels/taskinstance.py b/airflow-core/src/airflow/api_fastapi/execution_api/datamodels/taskinstance.py
index 8708ed51223..9f639b1e391 100644
--- a/airflow-core/src/airflow/api_fastapi/execution_api/datamodels/taskinstance.py
+++ b/airflow-core/src/airflow/api_fastapi/execution_api/datamodels/taskinstance.py
@@ -299,6 +299,7 @@ class DagRun(StrictBaseModel):
run_type: DagRunType
state: DagRunState
conf: dict[str, Any] | None = None
+ triggering_user_name: str | None = None
consumed_asset_events: list[AssetEventDagRunReference]
diff --git a/airflow-core/src/airflow/api_fastapi/execution_api/versions/__init__.py b/airflow-core/src/airflow/api_fastapi/execution_api/versions/__init__.py
index efe9710c9ec..a111b4dc91e 100644
--- a/airflow-core/src/airflow/api_fastapi/execution_api/versions/__init__.py
+++ b/airflow-core/src/airflow/api_fastapi/execution_api/versions/__init__.py
@@ -29,10 +29,12 @@ from airflow.api_fastapi.execution_api.versions.v2025_09_23 import AddDagVersion
from airflow.api_fastapi.execution_api.versions.v2025_10_27 import (
MakeDagRunConfNullable,
)
+from airflow.api_fastapi.execution_api.versions.v2025_11_05 import AddTriggeringUserNameField
bundle = VersionBundle(
HeadVersion(),
- Version("2025-10-27", MakeDagRunConfNullable),
+ Version("2025-11-05", MakeDagRunConfNullable),
+ Version("2025-10-10", AddTriggeringUserNameField),
Version("2025-09-23", AddDagVersionIdField),
Version(
"2025-08-10",
diff --git a/airflow-core/src/airflow/api_fastapi/execution_api/versions/v2025_11_05.py b/airflow-core/src/airflow/api_fastapi/execution_api/versions/v2025_11_05.py
new file mode 100644
index 00000000000..ced2917afda
--- /dev/null
+++ b/airflow-core/src/airflow/api_fastapi/execution_api/versions/v2025_11_05.py
@@ -0,0 +1,36 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from __future__ import annotations
+
+from cadwyn import ResponseInfo, VersionChange, convert_response_to_previous_version_for, schema
+
+from airflow.api_fastapi.execution_api.datamodels.taskinstance import DagRun, TIRunContext
+
+
+class AddTriggeringUserNameField(VersionChange):
+ """Add the `triggering_user_name` field to DagRun model."""
+
+ description = __doc__
+
+ instructions_to_migrate_to_previous_version = (schema(DagRun).field("triggering_user_name").didnt_exist,)
+
+ @convert_response_to_previous_version_for(TIRunContext) # type: ignore[arg-type]
+ def remove_triggering_user_name_from_dag_run(response: ResponseInfo) -> None: # type: ignore[misc]
+ """Remove the `triggering_user_name` field from the dag_run object when converting to the previous version."""
+ if "dag_run" in response.body and isinstance(response.body["dag_run"], dict):
+ response.body["dag_run"].pop("triggering_user_name", None)
diff --git a/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py b/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py
index a0e85de2a98..c12aee53bcc 100644
--- a/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py
+++ b/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py
@@ -189,6 +189,7 @@ class TestTIRunState:
"end_date": None,
"run_type": "manual",
"conf": {},
+ "triggering_user_name": None,
"consumed_asset_events": [],
},
"task_reschedule_count": 0,
@@ -637,6 +638,62 @@ class TestTIRunState:
assert response.status_code == 200
assert ti.xcom_pull(task_ids="test_xcom_not_cleared_for_deferral", key="key") == "value"
+ def test_ti_run_with_triggering_user_name(
+ self,
+ client,
+ session,
+ dag_maker,
+ time_machine,
+ ):
+ """
+ Test that the triggering_user_name field is correctly returned when it has a non-None value.
+ """
+ instant_str = "2024-09-30T12:00:00Z"
+ instant = timezone.parse(instant_str)
+ time_machine.move_to(instant, tick=False)
+
+ with dag_maker(dag_id=str(uuid4()), session=session):
+ EmptyOperator(task_id="test_ti_run_with_triggering_user_name")
+
+ # Create DagRun with triggering_user_name set to a specific value
+ dr = dag_maker.create_dagrun(
+ run_id="test",
+ logical_date=instant,
+ state=DagRunState.RUNNING,
+ start_date=instant,
+ triggering_user_name="test_user",
+ )
+
+ ti = dr.get_task_instance(task_id="test_ti_run_with_triggering_user_name")
+ ti.set_state(State.QUEUED)
+ session.commit()
+
+ response = client.patch(
+ f"/execution/task-instances/{ti.id}/run",
+ json={
+ "state": "running",
+ "hostname": "test-hostname",
+ "unixname": "test-unixname",
+ "pid": 12345,
+ "start_date": instant_str,
+ },
+ )
+
+ assert response.status_code == 200
+ json_response = response.json()
+
+ # Verify the dag_run is present
+ assert "dag_run" in json_response
+ dag_run = json_response["dag_run"]
+
+ # The triggering_user_name field should be present with the correct value
+ assert dag_run["triggering_user_name"] == "test_user"
+
+ # Verify other expected fields are still present
+ assert dag_run["dag_id"] == ti.dag_id
+ assert dag_run["run_id"] == "test"
+ assert dag_run["state"] == "running"
+
class TestTIUpdateState:
def setup_method(self):
diff --git a/airflow-core/tests/unit/api_fastapi/execution_api/versions/v2025_09_23/test_task_instances.py b/airflow-core/tests/unit/api_fastapi/execution_api/versions/v2025_09_23/test_task_instances.py
index 6f7af66f698..abca0f65adb 100644
--- a/airflow-core/tests/unit/api_fastapi/execution_api/versions/v2025_09_23/test_task_instances.py
+++ b/airflow-core/tests/unit/api_fastapi/execution_api/versions/v2025_09_23/test_task_instances.py
@@ -34,6 +34,71 @@ def ver_client(client):
return client
+class TestTIRunStateV20250923:
+ """Test that API version 2025-09-23 does NOT include triggering_user_name field."""
+
+ def setup_method(self):
+ clear_db_runs()
+
+ def teardown_method(self):
+ clear_db_runs()
+
+ def test_ti_run_excludes_triggering_user_name(
+ self,
+ ver_client,
+ session,
+ create_task_instance,
+ time_machine,
+ ):
+ """
+ Test that the triggering_user_name field is NOT present in API version 2025-09-23.
+
+ This field was added in version 2025-10-10, so older API clients should not
+ receive it in the response.
+ """
+ instant_str = "2024-09-30T12:00:00Z"
+ instant = timezone.parse(instant_str)
+ time_machine.move_to(instant, tick=False)
+
+ ti = create_task_instance(
+ task_id="test_triggering_user_exclusion",
+ state=State.QUEUED,
+ dagrun_state=DagRunState.RUNNING,
+ session=session,
+ start_date=instant,
+ )
+ session.commit()
+
+ response = ver_client.patch(
+ f"/execution/task-instances/{ti.id}/run",
+ json={
+ "state": "running",
+ "hostname": "test-hostname",
+ "unixname": "test-user",
+ "pid": 12345,
+ "start_date": instant_str,
+ },
+ )
+
+ assert response.status_code == 200
+ json_response = response.json()
+
+ # Verify the dag_run is present
+ assert "dag_run" in json_response
+ dag_run = json_response["dag_run"]
+
+ # The triggering_user_name field should NOT be present in this API version
+ assert "triggering_user_name" not in dag_run, (
+ "triggering_user_name should not be present in API version 2025-09-23"
+ )
+
+ # Verify other expected fields are still present
+ assert dag_run["dag_id"] == ti.dag_id
+ assert dag_run["run_id"] == "test"
+ assert dag_run["state"] == "running"
+ assert dag_run["conf"] == {}
+
+
class TestTIRunConfV20250923:
"""Test that API version 2025-09-23 converts NULL conf to empty dict."""
diff --git a/task-sdk/src/airflow/sdk/api/datamodels/_generated.py b/task-sdk/src/airflow/sdk/api/datamodels/_generated.py
index 4e9d2c6438c..d288fecc1f2 100644
--- a/task-sdk/src/airflow/sdk/api/datamodels/_generated.py
+++ b/task-sdk/src/airflow/sdk/api/datamodels/_generated.py
@@ -27,7 +27,7 @@ from uuid import UUID
from pydantic import AwareDatetime, BaseModel, ConfigDict, Field, JsonValue, RootModel
-API_VERSION: Final[str] = "2025-10-27"
+API_VERSION: Final[str] = "2025-11-05"
class AssetAliasReferenceAssetEventDagRun(BaseModel):
@@ -539,6 +539,7 @@ class DagRun(BaseModel):
run_type: DagRunType
state: DagRunState
conf: Annotated[dict[str, Any] | None, Field(title="Conf")] = None
+ triggering_user_name: Annotated[str | None, Field(title="Triggering User Name")] = None
consumed_asset_events: Annotated[list[AssetEventDagRunReference], Field(title="Consumed Asset Events")]
diff --git a/task-sdk/src/airflow/sdk/types.py b/task-sdk/src/airflow/sdk/types.py
index c7084629085..124fb560cc5 100644
--- a/task-sdk/src/airflow/sdk/types.py
+++ b/task-sdk/src/airflow/sdk/types.py
@@ -52,6 +52,7 @@ class DagRunProtocol(Protocol):
run_type: Any
run_after: AwareDatetime
conf: dict[str, Any] | None
+ triggering_user_name: str | None
class RuntimeTaskInstanceProtocol(Protocol):
diff --git a/task-sdk/tests/task_sdk/execution_time/test_supervisor.py b/task-sdk/tests/task_sdk/execution_time/test_supervisor.py
index bfd8caa3b47..6001edf737f 100644
--- a/task-sdk/tests/task_sdk/execution_time/test_supervisor.py
+++ b/task-sdk/tests/task_sdk/execution_time/test_supervisor.py
@@ -1847,6 +1847,7 @@ REQUEST_TEST_CASES = [
"end_date": None,
"clear_number": 0,
"conf": None,
+ "triggering_user_name": None,
},
"type": "PreviousDagRunResult",
},
@@ -1867,6 +1868,7 @@ REQUEST_TEST_CASES = [
run_after=timezone.parse("2024-01-15T12:00:00Z"),
consumed_asset_events=[],
state=DagRunState.SUCCESS,
+ triggering_user_name=None,
)
),
),
(airflow) 01/02: Reduce log level for Connection not found event from error to debug in task-sdk. (#57548)
Posted by ka...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
kaxilnaik pushed a commit to branch v3-1-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 860aa6f8ab8d6a1c9e23ce8c001dcf71b1cc7245
Author: Nitochkin <62...@users.noreply.github.com>
AuthorDate: Fri Oct 31 15:30:05 2025 +0100
Reduce log level for Connection not found event from error to debug in task-sdk. (#57548)
Co-authored-by: Anton Nitochkin <ni...@google.com>
(cherry picked from commit 56106ddb343d1948d71fce1d4f16d9d8c12f112b)
---
task-sdk/src/airflow/sdk/api/client.py | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/task-sdk/src/airflow/sdk/api/client.py b/task-sdk/src/airflow/sdk/api/client.py
index f4476322e8a..c4441dbdebd 100644
--- a/task-sdk/src/airflow/sdk/api/client.py
+++ b/task-sdk/src/airflow/sdk/api/client.py
@@ -356,7 +356,7 @@ class ConnectionOperations:
resp = self.client.get(f"connections/{conn_id}")
except ServerResponseError as e:
if e.response.status_code == HTTPStatus.NOT_FOUND:
- log.error(
+ log.debug(
"Connection not found",
conn_id=conn_id,
detail=e.detail,