You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "choo121600 (via GitHub)" <gi...@apache.org> on 2025/07/01 19:34:23 UTC
[PR] Fix inconsistent queued DAG filtering between dashboard and list page [airflow]
choo121600 opened a new pull request, #52668:
URL: https://github.com/apache/airflow/pull/52668
## Summary
This pull request fixes an inconsistency where Dags with queued DAG runs were correctly counted in the dashboard stats, but not shown in the Dag list page when using the last_dag_run_state=queued filter.
## Why?
When users clicked on the "Queued Dags" stat in the dashboard, they were redirected to the Dag list view, but the page displayed "No Dag found" despite the presence of queued Dags.
This resulted in a confusing and inconsistent user experience between the dashboard and the list view.
https://github.com/user-attachments/assets/bfd1ecc9-274b-4fa4-99ac-147ce754924d
## Additional context
The root cause appears to be related to Dags that are not "activated" in the UI — specifically, Dags with start_date = null (i.e., never scheduled), but were manually triggered and entered the queued state.
While these Dags have valid Dag runs in the queued state and are counted in the dashboard, the Dag list page excludes them due to how the filter is applied.
This PR adjusts the filtering logic so that the Dag list accurately reflects the Dags shown in the dashboard stats.
https://github.com/user-attachments/assets/58f6e70b-6b18-4ccc-8bf1-a717ca707896
closes: #52655
<!--
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.
-->
<!--
Thank you for contributing! Please make sure that your code changes
are covered with tests. And in case of new features or big changes
remember to adjust the documentation.
Feel free to ping committers for the review!
In case of an existing issue, reference it using one of the following:
closes: #ISSUE
related: #ISSUE
How to write a good git commit message:
http://chris.beams.io/posts/git-commit/
-->
<!-- Please keep an empty line above the dashes. -->
---
**^ Add meaningful description above**
Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)** for more information.
In case of fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed.
In case of a new dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
In case of backwards incompatible changes please leave a note in a newsfragment file, named `{pr_number}.significant.rst` or `{issue_number}.significant.rst`, in [airflow-core/newsfragments](https://github.com/apache/airflow/tree/main/airflow-core/newsfragments).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix Dag list filtering to include QUEUED DagRuns with null start_date [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on PR #52668:
URL: https://github.com/apache/airflow/pull/52668#issuecomment-3045577864
Whole bunch of commits to backport to be able to get that in, marking for 3.1.0.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix Dag list filtering to include QUEUED DagRuns with null start_date [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun merged PR #52668:
URL: https://github.com/apache/airflow/pull/52668
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix Dag list filtering to include QUEUED DagRuns with null start_date [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #52668:
URL: https://github.com/apache/airflow/pull/52668#discussion_r2190059652
##########
airflow-core/tests/unit/api_fastapi/common/db/test_dags.py:
##########
@@ -0,0 +1,273 @@
+# 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 datetime import datetime, timezone
+
+import pytest
+
+from airflow.api_fastapi.common.db.dags import generate_dag_with_latest_run_query
+from airflow.api_fastapi.common.parameters import SortParam
+from airflow.models import DagModel
+from airflow.models.dagrun import DagRun
+from airflow.utils.session import provide_session
+from airflow.utils.state import DagRunState
+from airflow.utils.timezone import utcnow
+
+from tests_common.test_utils.db import clear_db_dags, clear_db_runs
+
+pytestmark = pytest.mark.db_test
+
+
+class TestGenerateDagWithLatestRunQuery:
+ """Unit tests for generate_dag_with_latest_run_query function."""
+
+ @staticmethod
+ def _clear_db():
+ clear_db_runs()
+ clear_db_dags()
+
+ @pytest.fixture(autouse=True)
+ def setup_teardown(self):
+ """Setup and teardown for each test."""
+ self._clear_db()
+ yield
+ self._clear_db()
+
+ @pytest.fixture
+ @provide_session
+ def dag_with_queued_run(self, session):
+ """Returns a DAG with a QUEUED DagRun and null start_date."""
+
+ dag_id = "dag_with_queued_run"
+
+ # Create DagModel
+ dag_model = DagModel(
+ dag_id=dag_id,
+ is_stale=False,
+ is_paused=False,
+ fileloc="/tmp/dag.py",
+ )
+ session.add(dag_model)
+ session.flush()
+
+ # Create DagRun with start_date=None (QUEUED state)
+ dagrun = DagRun(
+ dag_id=dag_id,
+ run_id="manual__queued",
+ run_type="manual",
+ logical_date=utcnow(),
+ state=DagRunState.QUEUED,
+ start_date=None,
+ )
+ session.add(dagrun)
+ session.commit()
+
+ return dag_model, dagrun
+
+ @pytest.fixture
+ @provide_session
+ def dag_with_running_run(self, session):
+ """Returns a DAG with a RUNNING DagRun and a valid start_date."""
+
+ dag_id = "dag_with_running_run"
+
+ # Create DagModel
+ dag_model = DagModel(
+ dag_id=dag_id,
+ is_stale=False,
+ is_paused=False,
+ fileloc="/tmp/dag2.py",
+ )
+ session.add(dag_model)
+ session.flush()
+
+ # Create DagRun with start_date set (RUNNING state)
+ start_time = utcnow()
+ dagrun = DagRun(
+ dag_id=dag_id,
+ run_id="manual__running",
+ run_type="manual",
+ logical_date=start_time,
+ state=DagRunState.RUNNING,
+ start_date=start_time,
+ )
+ session.add(dagrun)
+ session.commit()
+
+ return dag_model, dagrun
+
+ @provide_session
+ def test_includes_queued_run_without_start_date(self, dag_with_queued_run, session):
+ """DAGs with QUEUED runs and null start_date should be included when no filters are applied, and joined DagRun state must not be None."""
+ dag_model, _ = dag_with_queued_run
+ query = generate_dag_with_latest_run_query(
+ max_run_filters=[],
+ order_by=SortParam(allowed_attrs=["dag_id"], model=DagModel).set_value("dag_id"),
+ )
+
+ # Also fetch joined DagRun's state and start_date
+ extended_query = query.add_columns(DagRun.state, DagRun.start_date)
+ result = session.execute(extended_query).fetchall()
+ dag_row = next((row for row in result if row[0].dag_id == dag_model.dag_id), None)
+ assert dag_row is not None
+ dagrun_state = dag_row[1]
+ assert dagrun_state is not None, "Joined DagRun state must not be None"
+
+ @provide_session
+ def test_includes_queued_run_when_ordering_by_state(
+ self, dag_with_queued_run, dag_with_running_run, session
+ ):
+ """DAGs with QUEUED runs and null start_date, and RUNNING runs must all have joined DagRun info not None."""
+ queued_dag_model, _ = dag_with_queued_run
+ running_dag_model, _ = dag_with_running_run
+
+ query = generate_dag_with_latest_run_query(
+ max_run_filters=[],
+ order_by=SortParam(allowed_attrs=["last_run_state"], model=DagModel).set_value("last_run_state"),
+ )
+ extended_query = query.add_columns(DagRun.state, DagRun.start_date)
+ result = session.execute(extended_query).fetchall()
+
+ # QUEUED DAG
+ queued_row = next((row for row in result if row[0].dag_id == queued_dag_model.dag_id), None)
+ assert queued_row is not None
+ assert queued_row[1] is not None, "Joined DagRun state for QUEUED DAG must not be None"
+ # RUNNING DAG
+ running_row = next((row for row in result if row[0].dag_id == running_dag_model.dag_id), None)
+ assert running_row is not None
+ assert running_row[1] is not None, "Joined DagRun state for RUNNING DAG must not be None"
+ assert running_row[2] is not None, "Joined DagRun start_date for RUNNING DAG must not be None"
+
+ @provide_session
+ def test_includes_queued_run_when_ordering_by_start_date(
+ self, dag_with_queued_run, dag_with_running_run, session
+ ):
+ """DAGs with QUEUED runs and RUNNING runs must all have joined DagRun info not None when ordering by start_date."""
+ queued_dag_model, _ = dag_with_queued_run
+ running_dag_model, _ = dag_with_running_run
+
+ query = generate_dag_with_latest_run_query(
+ max_run_filters=[],
+ order_by=SortParam(allowed_attrs=["last_run_start_date"], model=DagModel).set_value(
+ "last_run_start_date"
+ ),
+ )
+ extended_query = query.add_columns(DagRun.state, DagRun.start_date)
+ result = session.execute(extended_query).fetchall()
+
+ # QUEUED DAG
+ queued_row = next((row for row in result if row[0].dag_id == queued_dag_model.dag_id), None)
+ assert queued_row is not None
+ assert queued_row[1] is not None, "Joined DagRun state for QUEUED DAG must not be None"
+ # RUNNING DAG
+ running_row = next((row for row in result if row[0].dag_id == running_dag_model.dag_id), None)
+ assert running_row is not None
+ assert running_row[1] is not None, "Joined DagRun state for RUNNING DAG must not be None"
+ assert running_row[2] is not None, "Joined DagRun start_date for RUNNING DAG must not be None"
+
+ @provide_session
Review Comment:
```suggestion
```
This is not necessary, `session` is a fixture and will be provided by the test suite.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix inconsistent queued DAG filtering between dashboard and list page [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #52668:
URL: https://github.com/apache/airflow/pull/52668#discussion_r2182887776
##########
airflow-core/src/airflow/api_fastapi/common/db/dags.py:
##########
@@ -37,7 +38,7 @@ def generate_dag_with_latest_run_query(max_run_filters: list[BaseParam], order_b
max_run_id_query = ( # ordering by id will not always be "latest run", but it's a simplifying assumption
select(DagRun.dag_id, func.max(DagRun.id).label("max_dag_run_id"))
- .where(DagRun.start_date.is_not(null()))
+ .where(or_(DagRun.start_date.is_not(null()), DagRun.state == DagRunState.QUEUED))
Review Comment:
Also we most likely want to add a test case for this scenario.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix inconsistent queued DAG filtering between dashboard and list page [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #52668:
URL: https://github.com/apache/airflow/pull/52668#discussion_r2184704222
##########
airflow-core/src/airflow/api_fastapi/common/db/dags.py:
##########
@@ -37,7 +38,7 @@ def generate_dag_with_latest_run_query(max_run_filters: list[BaseParam], order_b
max_run_id_query = ( # ordering by id will not always be "latest run", but it's a simplifying assumption
select(DagRun.dag_id, func.max(DagRun.id).label("max_dag_run_id"))
- .where(DagRun.start_date.is_not(null()))
+ .where(or_(DagRun.start_date.is_not(null()), DagRun.state == DagRunState.QUEUED))
Review Comment:
> This function is currently only used in the /dashboard stats endpoint, so initially I handled it by allowing only QUEUED runs (which are the only valid state with start_date=None).
What function are you talking about `generate_dag_with_latest_run_query` ? `generate_dag_with_latest_run_query` is not used in the stats endpoint but in many other dag listing endpoints.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix inconsistent queued DAG filtering between dashboard and list page [airflow]
Posted by "choo121600 (via GitHub)" <gi...@apache.org>.
choo121600 commented on code in PR #52668:
URL: https://github.com/apache/airflow/pull/52668#discussion_r2185547202
##########
airflow-core/src/airflow/api_fastapi/common/db/dags.py:
##########
@@ -37,7 +38,7 @@ def generate_dag_with_latest_run_query(max_run_filters: list[BaseParam], order_b
max_run_id_query = ( # ordering by id will not always be "latest run", but it's a simplifying assumption
select(DagRun.dag_id, func.max(DagRun.id).label("max_dag_run_id"))
- .where(DagRun.start_date.is_not(null()))
+ .where(or_(DagRun.start_date.is_not(null()), DagRun.state == DagRunState.QUEUED))
Review Comment:
You're absolutely right — I initially thought `generate_dag_with_latest_run_query` was only used in the stats endpoint, but after taking a closer look today, I realized it's used in many places where DAGs are listed and sorted.
It seems I didn’t fully grasp the scope of its usage earlier. Thanks for pointing it out!
I'll make sure to take that into account and create appropriate test cases to cover these scenarios.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix inconsistent queued DAG filtering between dashboard and list page [airflow]
Posted by "choo121600 (via GitHub)" <gi...@apache.org>.
choo121600 commented on code in PR #52668:
URL: https://github.com/apache/airflow/pull/52668#discussion_r2184115457
##########
airflow-core/src/airflow/api_fastapi/common/db/dags.py:
##########
@@ -37,7 +38,7 @@ def generate_dag_with_latest_run_query(max_run_filters: list[BaseParam], order_b
max_run_id_query = ( # ordering by id will not always be "latest run", but it's a simplifying assumption
select(DagRun.dag_id, func.max(DagRun.id).label("max_dag_run_id"))
- .where(DagRun.start_date.is_not(null()))
+ .where(or_(DagRun.start_date.is_not(null()), DagRun.state == DagRunState.QUEUED))
Review Comment:
Thanks for the feedback!
This function is currently only used in the /dashboard stats endpoint, so initially I handled it by allowing only QUEUED runs (which are the only valid state with start_date=None).
However, I checked other places like the Grid UI and found that they do not use start_date as a filter when retrieving the latest run. In fact, this function is the only one that includes .where(DagRun.start_date.is_not(null())), which seems inconsistent with the intended purpose of the function — returning the latest DagRun regardless of state.
So I agree — we should remove that line entirely.
I’ll go ahead and revise the PR accordingly and add a test case to cover this scenario.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix inconsistent queued DAG filtering between dashboard and list page [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #52668:
URL: https://github.com/apache/airflow/pull/52668#discussion_r2182878696
##########
airflow-core/src/airflow/api_fastapi/common/db/dags.py:
##########
@@ -37,7 +38,7 @@ def generate_dag_with_latest_run_query(max_run_filters: list[BaseParam], order_b
max_run_id_query = ( # ordering by id will not always be "latest run", but it's a simplifying assumption
select(DagRun.dag_id, func.max(DagRun.id).label("max_dag_run_id"))
- .where(DagRun.start_date.is_not(null()))
+ .where(or_(DagRun.start_date.is_not(null()), DagRun.state == DagRunState.QUEUED))
Review Comment:
I think this problem will be here for any state that has a `start_date` None. (no_status, scheduled, queued, etc....)
##########
airflow-core/src/airflow/api_fastapi/common/db/dags.py:
##########
@@ -37,7 +38,7 @@ def generate_dag_with_latest_run_query(max_run_filters: list[BaseParam], order_b
max_run_id_query = ( # ordering by id will not always be "latest run", but it's a simplifying assumption
select(DagRun.dag_id, func.max(DagRun.id).label("max_dag_run_id"))
- .where(DagRun.start_date.is_not(null()))
+ .where(or_(DagRun.start_date.is_not(null()), DagRun.state == DagRunState.QUEUED))
Review Comment:
Maybe we should completely remove that `where` clause. The run does not have to be 'started' to actually be considered, WDYT?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix Dag list filtering to include QUEUED DagRuns with null start_date [airflow]
Posted by "choo121600 (via GitHub)" <gi...@apache.org>.
choo121600 commented on code in PR #52668:
URL: https://github.com/apache/airflow/pull/52668#discussion_r2190220773
##########
airflow-core/tests/unit/api_fastapi/common/db/test_dags.py:
##########
@@ -0,0 +1,273 @@
+# 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 datetime import datetime, timezone
+
+import pytest
+
+from airflow.api_fastapi.common.db.dags import generate_dag_with_latest_run_query
+from airflow.api_fastapi.common.parameters import SortParam
+from airflow.models import DagModel
+from airflow.models.dagrun import DagRun
+from airflow.utils.session import provide_session
+from airflow.utils.state import DagRunState
+from airflow.utils.timezone import utcnow
+
+from tests_common.test_utils.db import clear_db_dags, clear_db_runs
+
+pytestmark = pytest.mark.db_test
+
+
+class TestGenerateDagWithLatestRunQuery:
+ """Unit tests for generate_dag_with_latest_run_query function."""
+
+ @staticmethod
+ def _clear_db():
+ clear_db_runs()
+ clear_db_dags()
+
+ @pytest.fixture(autouse=True)
+ def setup_teardown(self):
+ """Setup and teardown for each test."""
+ self._clear_db()
+ yield
+ self._clear_db()
+
+ @pytest.fixture
+ @provide_session
+ def dag_with_queued_run(self, session):
+ """Returns a DAG with a QUEUED DagRun and null start_date."""
+
+ dag_id = "dag_with_queued_run"
+
+ # Create DagModel
+ dag_model = DagModel(
+ dag_id=dag_id,
+ is_stale=False,
+ is_paused=False,
+ fileloc="/tmp/dag.py",
+ )
+ session.add(dag_model)
+ session.flush()
+
+ # Create DagRun with start_date=None (QUEUED state)
+ dagrun = DagRun(
+ dag_id=dag_id,
+ run_id="manual__queued",
+ run_type="manual",
+ logical_date=utcnow(),
+ state=DagRunState.QUEUED,
+ start_date=None,
+ )
+ session.add(dagrun)
+ session.commit()
+
+ return dag_model, dagrun
+
+ @pytest.fixture
+ @provide_session
+ def dag_with_running_run(self, session):
+ """Returns a DAG with a RUNNING DagRun and a valid start_date."""
+
+ dag_id = "dag_with_running_run"
+
+ # Create DagModel
+ dag_model = DagModel(
+ dag_id=dag_id,
+ is_stale=False,
+ is_paused=False,
+ fileloc="/tmp/dag2.py",
+ )
+ session.add(dag_model)
+ session.flush()
+
+ # Create DagRun with start_date set (RUNNING state)
+ start_time = utcnow()
+ dagrun = DagRun(
+ dag_id=dag_id,
+ run_id="manual__running",
+ run_type="manual",
+ logical_date=start_time,
+ state=DagRunState.RUNNING,
+ start_date=start_time,
+ )
+ session.add(dagrun)
+ session.commit()
+
+ return dag_model, dagrun
+
+ @provide_session
+ def test_includes_queued_run_without_start_date(self, dag_with_queued_run, session):
+ """DAGs with QUEUED runs and null start_date should be included when no filters are applied, and joined DagRun state must not be None."""
+ dag_model, _ = dag_with_queued_run
+ query = generate_dag_with_latest_run_query(
+ max_run_filters=[],
+ order_by=SortParam(allowed_attrs=["dag_id"], model=DagModel).set_value("dag_id"),
+ )
+
+ # Also fetch joined DagRun's state and start_date
+ extended_query = query.add_columns(DagRun.state, DagRun.start_date)
+ result = session.execute(extended_query).fetchall()
+ dag_row = next((row for row in result if row[0].dag_id == dag_model.dag_id), None)
+ assert dag_row is not None
+ dagrun_state = dag_row[1]
+ assert dagrun_state is not None, "Joined DagRun state must not be None"
+
+ @provide_session
+ def test_includes_queued_run_when_ordering_by_state(
+ self, dag_with_queued_run, dag_with_running_run, session
+ ):
+ """DAGs with QUEUED runs and null start_date, and RUNNING runs must all have joined DagRun info not None."""
+ queued_dag_model, _ = dag_with_queued_run
+ running_dag_model, _ = dag_with_running_run
+
+ query = generate_dag_with_latest_run_query(
+ max_run_filters=[],
+ order_by=SortParam(allowed_attrs=["last_run_state"], model=DagModel).set_value("last_run_state"),
+ )
+ extended_query = query.add_columns(DagRun.state, DagRun.start_date)
+ result = session.execute(extended_query).fetchall()
+
+ # QUEUED DAG
+ queued_row = next((row for row in result if row[0].dag_id == queued_dag_model.dag_id), None)
+ assert queued_row is not None
+ assert queued_row[1] is not None, "Joined DagRun state for QUEUED DAG must not be None"
+ # RUNNING DAG
+ running_row = next((row for row in result if row[0].dag_id == running_dag_model.dag_id), None)
+ assert running_row is not None
+ assert running_row[1] is not None, "Joined DagRun state for RUNNING DAG must not be None"
+ assert running_row[2] is not None, "Joined DagRun start_date for RUNNING DAG must not be None"
+
+ @provide_session
+ def test_includes_queued_run_when_ordering_by_start_date(
+ self, dag_with_queued_run, dag_with_running_run, session
+ ):
+ """DAGs with QUEUED runs and RUNNING runs must all have joined DagRun info not None when ordering by start_date."""
+ queued_dag_model, _ = dag_with_queued_run
+ running_dag_model, _ = dag_with_running_run
+
+ query = generate_dag_with_latest_run_query(
+ max_run_filters=[],
+ order_by=SortParam(allowed_attrs=["last_run_start_date"], model=DagModel).set_value(
+ "last_run_start_date"
+ ),
+ )
+ extended_query = query.add_columns(DagRun.state, DagRun.start_date)
+ result = session.execute(extended_query).fetchall()
+
+ # QUEUED DAG
+ queued_row = next((row for row in result if row[0].dag_id == queued_dag_model.dag_id), None)
+ assert queued_row is not None
+ assert queued_row[1] is not None, "Joined DagRun state for QUEUED DAG must not be None"
+ # RUNNING DAG
+ running_row = next((row for row in result if row[0].dag_id == running_dag_model.dag_id), None)
+ assert running_row is not None
+ assert running_row[1] is not None, "Joined DagRun state for RUNNING DAG must not be None"
+ assert running_row[2] is not None, "Joined DagRun start_date for RUNNING DAG must not be None"
+
+ @provide_session
Review Comment:
Thanks for catching that! I've removed the unnecessary `@provide_session` decorator😉
the session fixture handles it as expected👍
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix Dag list filtering to include QUEUED DagRuns with null start_date [airflow]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #52668:
URL: https://github.com/apache/airflow/pull/52668#issuecomment-3045534752
### Backport failed to create: v3-0-test. View the failure log <a href='https://github.com/apache/airflow/actions/runs/16120516646'> Run details </a>
<table>
<tr>
<th>Status</th>
<th>Branch</th>
<th>Result</th>
</tr>
<tr>
<td>❌</td>
<td>v3-0-test</td>
<td><a href="https://github.com/apache/airflow/commit/6f295a48b91772116c10aded6c269b4f075f8ba2"><img src='https://img.shields.io/badge/Commit-6f295a4-red' alt='Commit Link'></a></td>
</tr>
</table>
You can attempt to backport this manually by running:
```bash
cherry_picker 6f295a4 v3-0-test
```
This should apply the commit to the v3-0-test branch and leave the commit in conflict state marking
the files that need manual conflict resolution.
After you have resolved the conflicts, you can continue the backport process by running:
```bash
cherry_picker --continue
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org