You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by ep...@apache.org on 2023/05/08 16:39:10 UTC
[airflow] 17/37: Fix `order_by` request in list DAG rest api (#30926)
This is an automated email from the ASF dual-hosted git repository.
ephraimanierobi pushed a commit to branch v2-6-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit ddecc0fb9d0e28eff7a5ff0cc4113037718acad1
Author: Pankaj Singh <98...@users.noreply.github.com>
AuthorDate: Sat Apr 29 21:37:01 2023 +0530
Fix `order_by` request in list DAG rest api (#30926)
(cherry picked from commit 1d4b1410b027c667d4e2f51f488f98b166facf71)
---
airflow/api_connexion/endpoints/dag_endpoint.py | 8 ++-
tests/api_connexion/endpoints/test_dag_endpoint.py | 80 ++++++++++++++++++++++
2 files changed, 85 insertions(+), 3 deletions(-)
diff --git a/airflow/api_connexion/endpoints/dag_endpoint.py b/airflow/api_connexion/endpoints/dag_endpoint.py
index 83d861ba65..18dcb45be6 100644
--- a/airflow/api_connexion/endpoints/dag_endpoint.py
+++ b/airflow/api_connexion/endpoints/dag_endpoint.py
@@ -28,7 +28,7 @@ from sqlalchemy.sql.expression import or_
from airflow import DAG
from airflow.api_connexion import security
from airflow.api_connexion.exceptions import AlreadyExists, BadRequest, NotFound
-from airflow.api_connexion.parameters import check_limit, format_parameters
+from airflow.api_connexion.parameters import apply_sorting, check_limit, format_parameters
from airflow.api_connexion.schemas.dag_schema import (
DAGCollection,
dag_detail_schema,
@@ -75,9 +75,11 @@ def get_dags(
dag_id_pattern: str | None = None,
only_active: bool = True,
paused: bool | None = None,
+ order_by: str = "dag_id",
session: Session = NEW_SESSION,
) -> APIResponse:
"""Get all DAGs."""
+ allowed_attrs = ["dag_id"]
dags_query = session.query(DagModel).filter(~DagModel.is_subdag)
if only_active:
dags_query = dags_query.filter(DagModel.is_active)
@@ -97,8 +99,8 @@ def get_dags(
dags_query = dags_query.filter(or_(*cond))
total_entries = dags_query.count()
-
- dags = dags_query.order_by(DagModel.dag_id).offset(offset).limit(limit).all()
+ dags_query = apply_sorting(dags_query, order_by, {}, allowed_attrs)
+ dags = dags_query.offset(offset).limit(limit).all()
return dags_collection_schema.dump(DAGCollection(dags=dags, total_entries=total_entries))
diff --git a/tests/api_connexion/endpoints/test_dag_endpoint.py b/tests/api_connexion/endpoints/test_dag_endpoint.py
index 2298c58d80..5d0542a2c6 100644
--- a/tests/api_connexion/endpoints/test_dag_endpoint.py
+++ b/tests/api_connexion/endpoints/test_dag_endpoint.py
@@ -1722,6 +1722,86 @@ class TestPatchDags(TestDagEndpoint):
dags_updated = session.query(DagModel).filter(DagModel.is_paused)
assert len(dags_updated.all()) == 2
+ @provide_session
+ def test_should_respond_200_and_reverse_ordering(self, session, url_safe_serializer):
+ file_token = url_safe_serializer.dumps("/tmp/dag_1.py")
+ self._create_dag_models(2)
+ file_token10 = url_safe_serializer.dumps("/tmp/dag_2.py")
+
+ response = self.client.get(
+ "/api/v1/dags?order_by=-dag_id",
+ environ_overrides={"REMOTE_USER": "test"},
+ )
+
+ assert response.status_code == 200
+ assert {
+ "dags": [
+ {
+ "dag_id": "TEST_DAG_2",
+ "description": None,
+ "fileloc": "/tmp/dag_2.py",
+ "file_token": file_token10,
+ "is_paused": False,
+ "is_active": True,
+ "is_subdag": False,
+ "owners": [],
+ "root_dag_id": None,
+ "schedule_interval": {
+ "__type": "CronExpression",
+ "value": "2 2 * * *",
+ },
+ "tags": [],
+ "next_dagrun": None,
+ "has_task_concurrency_limits": True,
+ "next_dagrun_data_interval_start": None,
+ "next_dagrun_data_interval_end": None,
+ "max_active_runs": 16,
+ "next_dagrun_create_after": None,
+ "last_expired": None,
+ "max_active_tasks": 16,
+ "last_pickled": None,
+ "default_view": None,
+ "last_parsed_time": None,
+ "scheduler_lock": None,
+ "timetable_description": None,
+ "has_import_errors": False,
+ "pickle_id": None,
+ },
+ {
+ "dag_id": "TEST_DAG_1",
+ "description": None,
+ "fileloc": "/tmp/dag_1.py",
+ "file_token": file_token,
+ "is_paused": False,
+ "is_active": True,
+ "is_subdag": False,
+ "owners": [],
+ "root_dag_id": None,
+ "schedule_interval": {
+ "__type": "CronExpression",
+ "value": "2 2 * * *",
+ },
+ "tags": [],
+ "next_dagrun": None,
+ "has_task_concurrency_limits": True,
+ "next_dagrun_data_interval_start": None,
+ "next_dagrun_data_interval_end": None,
+ "max_active_runs": 16,
+ "next_dagrun_create_after": None,
+ "last_expired": None,
+ "max_active_tasks": 16,
+ "last_pickled": None,
+ "default_view": None,
+ "last_parsed_time": None,
+ "scheduler_lock": None,
+ "timetable_description": None,
+ "has_import_errors": False,
+ "pickle_id": None,
+ },
+ ],
+ "total_entries": 2,
+ } == response.json
+
def test_should_respons_400_dag_id_pattern_missing(self):
self._create_dag_models(1)
response = self.client.patch(