This is an automated email from the ASF dual-hosted git repository.
amoghrajesh 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 403a1ebaa77 Fix task run context crash when DagRun state is expired
(#66339)
403a1ebaa77 is described below
commit 403a1ebaa77c05a398004e94106733bb35252954
Author: Amogh Desai <[email protected]>
AuthorDate: Mon May 4 15:47:32 2026 +0530
Fix task run context crash when DagRun state is expired (#66339)
---
.../execution_api/datamodels/taskinstance.py | 7 +++++--
.../tests/unit/cli/commands/test_task_command.py | 17 +++++++++++++++++
2 files changed, 22 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 adef323565f..a0d97390801 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
@@ -338,8 +338,11 @@ class DagRun(StrictBaseModel):
for field_name in cls.model_fields:
if field_name in insp.dict:
values[field_name] = insp.dict[field_name]
- elif field_name == "state" and "_state" in insp.dict:
- values["state"] = insp.dict["_state"]
+ elif field_name == "state":
+ if "_state" in insp.dict:
+ values["state"] = insp.dict["_state"]
+ elif not insp.detached and (state_val := data._state) is not
None:
+ values["state"] = state_val
if "consumed_asset_events" not in values:
values["consumed_asset_events"] = []
diff --git a/airflow-core/tests/unit/cli/commands/test_task_command.py
b/airflow-core/tests/unit/cli/commands/test_task_command.py
index b66384d5ad4..f39a8409bc2 100644
--- a/airflow-core/tests/unit/cli/commands/test_task_command.py
+++ b/airflow-core/tests/unit/cli/commands/test_task_command.py
@@ -307,6 +307,23 @@ class TestCliTasks:
with redirect_stdout(io.StringIO()):
task_command.task_render(args)
+ @pytest.mark.db_test
+ def test_task_render_handles_expired_dagrun(self, dag_maker, session):
+ """Test that model_validate extracts state from an expired DagRun
instance."""
+ from airflow.api_fastapi.execution_api.datamodels.taskinstance import
DagRun as DagRunPydantic
+ from airflow.utils.state import DagRunState
+
+ with dag_maker(dag_id="test_expired", session=session):
+ pass
+
+ dr = dag_maker.create_dagrun(state=DagRunState.RUNNING)
+ session.commit()
+ # After commit, SQLAlchemy expires all attributes — _state is no
longer in insp.dict
+ # but the instance is still attached, so direct access triggers a lazy
reload.
+
+ pydantic_dr = DagRunPydantic.model_validate(dr)
+ assert pydantic_dr.state == DagRunState.RUNNING
+
@pytest.mark.usefixtures("testing_dag_bundle")
def test_mapped_task_render(self):
"""
