This is an automated email from the ASF dual-hosted git repository.
henry3260 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 d3972ae1f6e Simplify airflowctl query parameter handling (#69475)
d3972ae1f6e is described below
commit d3972ae1f6ef4f6b1b9137a495a8fc0a965ad0ea
Author: Henry Chen <[email protected]>
AuthorDate: Wed Jul 8 01:57:10 2026 +0800
Simplify airflowctl query parameter handling (#69475)
---
airflow-ctl/src/airflowctl/api/operations.py | 40 +++++++++++-----------
.../tests/airflow_ctl/api/test_operations.py | 16 ++++++++-
2 files changed, 35 insertions(+), 21 deletions(-)
diff --git a/airflow-ctl/src/airflowctl/api/operations.py
b/airflow-ctl/src/airflowctl/api/operations.py
index ad0de6d008b..f58ad78adeb 100644
--- a/airflow-ctl/src/airflowctl/api/operations.py
+++ b/airflow-ctl/src/airflowctl/api/operations.py
@@ -88,6 +88,16 @@ log = structlog.get_logger(logger_name=__name__)
T = TypeVar("T", bound=BaseModel)
+def _serialize_query_param(value: Any) -> Any:
+ if isinstance(value, datetime.datetime):
+ return value.isoformat()
+ return value
+
+
+def _build_query_params(**values: Any) -> dict[str, Any]:
+ return {name: _serialize_query_param(value) for name, value in
values.items() if value is not None}
+
+
# Generic Server Response Error
class ServerResponseError(httpx.HTTPStatusError):
"""Server response error (Generic)."""
@@ -642,19 +652,15 @@ class DagRunOperations(BaseOperations):
if not dag_id:
dag_id = "~"
- params: dict[str, Any] = {"limit": limit}
- if state is not None:
- params["state"] = str(state)
- if start_date is not None:
- params["start_date"] = start_date.isoformat()
- if end_date is not None:
- params["end_date"] = end_date.isoformat()
- if logical_date_gte is not None:
- params["logical_date_gte"] = logical_date_gte.isoformat()
- if logical_date_lte is not None:
- params["logical_date_lte"] = logical_date_lte.isoformat()
- if order_by is not None:
- params["order_by"] = order_by
+ params = _build_query_params(
+ limit=limit,
+ state=str(state) if state is not None else None,
+ start_date=start_date,
+ end_date=end_date,
+ logical_date_gte=logical_date_gte,
+ logical_date_lte=logical_date_lte,
+ order_by=order_by,
+ )
try:
self.response = self.client.get(f"/dags/{dag_id}/dagRuns",
params=params)
@@ -681,13 +687,7 @@ class JobsOperations(BaseOperations):
is_alive: bool | None = None,
) -> JobCollectionResponse | ServerResponseError:
"""List all jobs."""
- params: dict[str, Any] = {}
- if job_type:
- params["job_type"] = job_type
- if hostname:
- params["hostname"] = hostname
- if is_alive is not None:
- params["is_alive"] = is_alive
+ params = _build_query_params(job_type=job_type or None,
hostname=hostname or None, is_alive=is_alive)
return super().execute_list(path="jobs",
data_model=JobCollectionResponse, params=params)
diff --git a/airflow-ctl/tests/airflow_ctl/api/test_operations.py
b/airflow-ctl/tests/airflow_ctl/api/test_operations.py
index 53dfb321717..dc5cdd396db 100644
--- a/airflow-ctl/tests/airflow_ctl/api/test_operations.py
+++ b/airflow-ctl/tests/airflow_ctl/api/test_operations.py
@@ -100,7 +100,7 @@ from airflowctl.api.datamodels.generated import (
XComResponse,
XComResponseNative,
)
-from airflowctl.api.operations import BaseOperations
+from airflowctl.api.operations import BaseOperations, _build_query_params
from airflowctl.exceptions import AirflowCtlConnectionException
if TYPE_CHECKING:
@@ -127,6 +127,20 @@ class HelloCollectionResponse(BaseModel):
class TestBaseOperations:
+ def test_build_query_params_skips_none_and_serializes_datetime(self):
+ logical_date = datetime.datetime(2025, 1, 1, 12, 30,
tzinfo=datetime.timezone.utc)
+
+ assert _build_query_params(
+ logical_date=logical_date,
+ state=None,
+ limit=10,
+ order_by="-id",
+ ) == {
+ "logical_date": logical_date.isoformat(),
+ "limit": 10,
+ "order_by": "-id",
+ }
+
def test_server_connection_refused(self):
client = make_api_client(base_url="http://localhost";)
with pytest.raises(
