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 5227a911412 [AIP-94] Add filters and pagination to airflowctl jobs
list (#68616)
5227a911412 is described below
commit 5227a911412d925e6d892be68a5f6f0a4cd33e0a
Author: Henry Chen <[email protected]>
AuthorDate: Tue Jul 21 01:31:54 2026 +0800
[AIP-94] Add filters and pagination to airflowctl jobs list (#68616)
* Support legacy list-jobs filters in airflowctl jobs list
* Always default jobs list ordering to most recent start date
Previously the default order_by switched between the API default and
-start_date depending on whether --limit was given, making the output
ordering hard to predict. The legacy airflow dags list-jobs command
always ordered by start_date descending, so apply that default
unconditionally.
Co-Authored-By: Claude Fable 5 <[email protected]>
---------
Co-authored-by: Claude Fable 5 <[email protected]>
---
.../core_api/openapi/v2-rest-api-generated.yaml | 8 +++
.../api_fastapi/core_api/routes/public/job.py | 4 ++
.../src/airflow/ui/openapi-gen/queries/common.ts | 5 +-
.../ui/openapi-gen/queries/ensureQueryData.ts | 6 +-
.../src/airflow/ui/openapi-gen/queries/prefetch.ts | 6 +-
.../src/airflow/ui/openapi-gen/queries/queries.ts | 6 +-
.../src/airflow/ui/openapi-gen/queries/suspense.ts | 6 +-
.../ui/openapi-gen/requests/services.gen.ts | 2 +
.../airflow/ui/openapi-gen/requests/types.gen.ts | 1 +
.../api_fastapi/core_api/routes/public/test_job.py | 17 ++++++
airflow-ctl/src/airflowctl/api/operations.py | 20 ++++++-
.../tests/airflow_ctl/api/test_operations.py | 68 +++++++++++++++++++---
12 files changed, 131 insertions(+), 18 deletions(-)
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
b/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
index d1a6ee83fd0..1b32b764372 100644
---
a/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
+++
b/airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml
@@ -5404,6 +5404,14 @@ paths:
- type: string
- type: 'null'
title: Job State
+ - name: dag_id
+ in: query
+ required: false
+ schema:
+ anyOf:
+ - type: string
+ - type: 'null'
+ title: Dag Id
- name: job_type
in: query
required: false
diff --git a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/job.py
b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/job.py
index b56c78f2de6..3e7e16bb69a 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/job.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/job.py
@@ -86,6 +86,9 @@ def get_jobs(
state: Annotated[
FilterParam[str | None], Depends(filter_param_factory(Job.state, str |
None, filter_name="job_state"))
],
+ dag_id: Annotated[
+ FilterParam[str | None], Depends(filter_param_factory(Job.dag_id, str
| None, filter_name="dag_id"))
+ ],
job_type: Annotated[
FilterParam[str | None],
Depends(filter_param_factory(Job.job_type, str | None,
filter_name="job_type")),
@@ -109,6 +112,7 @@ def get_jobs(
start_date_range,
end_date_range,
state,
+ dag_id,
job_type,
hostname,
executor_class,
diff --git a/airflow-core/src/airflow/ui/openapi-gen/queries/common.ts
b/airflow-core/src/airflow/ui/openapi-gen/queries/common.ts
index fa4853f6050..286a724232e 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/common.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/common.ts
@@ -703,7 +703,8 @@ export const UseImportErrorServiceGetImportErrorsKeyFn = ({
bundleName, filename
export type JobServiceGetJobsDefaultResponse = Awaited<ReturnType<typeof
JobService.getJobs>>;
export type JobServiceGetJobsQueryResult<TData =
JobServiceGetJobsDefaultResponse, TError = unknown> = UseQueryResult<TData,
TError>;
export const useJobServiceGetJobsKey = "JobServiceGetJobs";
-export const UseJobServiceGetJobsKeyFn = ({ endDateGt, endDateGte, endDateLt,
endDateLte, executorClass, hostname, isAlive, jobState, jobType, limit, offset,
orderBy, startDateGt, startDateGte, startDateLt, startDateLte }: {
+export const UseJobServiceGetJobsKeyFn = ({ dagId, endDateGt, endDateGte,
endDateLt, endDateLte, executorClass, hostname, isAlive, jobState, jobType,
limit, offset, orderBy, startDateGt, startDateGte, startDateLt, startDateLte }:
{
+ dagId?: string;
endDateGt?: string;
endDateGte?: string;
endDateLt?: string;
@@ -720,7 +721,7 @@ export const UseJobServiceGetJobsKeyFn = ({ endDateGt,
endDateGte, endDateLt, en
startDateGte?: string;
startDateLt?: string;
startDateLte?: string;
-} = {}, queryKey?: Array<unknown>) => [useJobServiceGetJobsKey, ...(queryKey
?? [{ endDateGt, endDateGte, endDateLt, endDateLte, executorClass, hostname,
isAlive, jobState, jobType, limit, offset, orderBy, startDateGt, startDateGte,
startDateLt, startDateLte }])];
+} = {}, queryKey?: Array<unknown>) => [useJobServiceGetJobsKey, ...(queryKey
?? [{ dagId, endDateGt, endDateGte, endDateLt, endDateLte, executorClass,
hostname, isAlive, jobState, jobType, limit, offset, orderBy, startDateGt,
startDateGte, startDateLt, startDateLte }])];
export type PluginServiceGetPluginsDefaultResponse = Awaited<ReturnType<typeof
PluginService.getPlugins>>;
export type PluginServiceGetPluginsQueryResult<TData =
PluginServiceGetPluginsDefaultResponse, TError = unknown> =
UseQueryResult<TData, TError>;
export const usePluginServiceGetPluginsKey = "PluginServiceGetPlugins";
diff --git a/airflow-core/src/airflow/ui/openapi-gen/queries/ensureQueryData.ts
b/airflow-core/src/airflow/ui/openapi-gen/queries/ensureQueryData.ts
index f83894f1bef..5984e9b22f6 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/ensureQueryData.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/ensureQueryData.ts
@@ -1441,13 +1441,15 @@ export const
ensureUseImportErrorServiceGetImportErrorsData = (queryClient: Quer
* @param data.offset
* @param data.orderBy Attributes to order by, multi criteria sort is
supported. Prefix with `-` for descending order. Supported attributes: `id,
dag_id, state, job_type, start_date, end_date, latest_heartbeat,
executor_class, hostname, unixname`
* @param data.jobState
+* @param data.dagId
* @param data.jobType
* @param data.hostname
* @param data.executorClass
* @returns JobCollectionResponse Successful Response
* @throws ApiError
*/
-export const ensureUseJobServiceGetJobsData = (queryClient: QueryClient, {
endDateGt, endDateGte, endDateLt, endDateLte, executorClass, hostname, isAlive,
jobState, jobType, limit, offset, orderBy, startDateGt, startDateGte,
startDateLt, startDateLte }: {
+export const ensureUseJobServiceGetJobsData = (queryClient: QueryClient, {
dagId, endDateGt, endDateGte, endDateLt, endDateLte, executorClass, hostname,
isAlive, jobState, jobType, limit, offset, orderBy, startDateGt, startDateGte,
startDateLt, startDateLte }: {
+ dagId?: string;
endDateGt?: string;
endDateGte?: string;
endDateLt?: string;
@@ -1464,7 +1466,7 @@ export const ensureUseJobServiceGetJobsData =
(queryClient: QueryClient, { endDa
startDateGte?: string;
startDateLt?: string;
startDateLte?: string;
-} = {}) => queryClient.ensureQueryData({ queryKey:
Common.UseJobServiceGetJobsKeyFn({ endDateGt, endDateGte, endDateLt,
endDateLte, executorClass, hostname, isAlive, jobState, jobType, limit, offset,
orderBy, startDateGt, startDateGte, startDateLt, startDateLte }), queryFn: ()
=> JobService.getJobs({ endDateGt, endDateGte, endDateLt, endDateLte,
executorClass, hostname, isAlive, jobState, jobType, limit, offset, orderBy,
startDateGt, startDateGte, startDateLt, startDateLte }) });
+} = {}) => queryClient.ensureQueryData({ queryKey:
Common.UseJobServiceGetJobsKeyFn({ dagId, endDateGt, endDateGte, endDateLt,
endDateLte, executorClass, hostname, isAlive, jobState, jobType, limit, offset,
orderBy, startDateGt, startDateGte, startDateLt, startDateLte }), queryFn: ()
=> JobService.getJobs({ dagId, endDateGt, endDateGte, endDateLt, endDateLte,
executorClass, hostname, isAlive, jobState, jobType, limit, offset, orderBy,
startDateGt, startDateGte, startDateLt, startDateLte }) });
/**
* Get Plugins
* @param data The data for the request.
diff --git a/airflow-core/src/airflow/ui/openapi-gen/queries/prefetch.ts
b/airflow-core/src/airflow/ui/openapi-gen/queries/prefetch.ts
index 1b4b2d46aac..cca9470844f 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/prefetch.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/prefetch.ts
@@ -1441,13 +1441,15 @@ export const
prefetchUseImportErrorServiceGetImportErrors = (queryClient: QueryC
* @param data.offset
* @param data.orderBy Attributes to order by, multi criteria sort is
supported. Prefix with `-` for descending order. Supported attributes: `id,
dag_id, state, job_type, start_date, end_date, latest_heartbeat,
executor_class, hostname, unixname`
* @param data.jobState
+* @param data.dagId
* @param data.jobType
* @param data.hostname
* @param data.executorClass
* @returns JobCollectionResponse Successful Response
* @throws ApiError
*/
-export const prefetchUseJobServiceGetJobs = (queryClient: QueryClient, {
endDateGt, endDateGte, endDateLt, endDateLte, executorClass, hostname, isAlive,
jobState, jobType, limit, offset, orderBy, startDateGt, startDateGte,
startDateLt, startDateLte }: {
+export const prefetchUseJobServiceGetJobs = (queryClient: QueryClient, {
dagId, endDateGt, endDateGte, endDateLt, endDateLte, executorClass, hostname,
isAlive, jobState, jobType, limit, offset, orderBy, startDateGt, startDateGte,
startDateLt, startDateLte }: {
+ dagId?: string;
endDateGt?: string;
endDateGte?: string;
endDateLt?: string;
@@ -1464,7 +1466,7 @@ export const prefetchUseJobServiceGetJobs = (queryClient:
QueryClient, { endDate
startDateGte?: string;
startDateLt?: string;
startDateLte?: string;
-} = {}) => queryClient.prefetchQuery({ queryKey:
Common.UseJobServiceGetJobsKeyFn({ endDateGt, endDateGte, endDateLt,
endDateLte, executorClass, hostname, isAlive, jobState, jobType, limit, offset,
orderBy, startDateGt, startDateGte, startDateLt, startDateLte }), queryFn: ()
=> JobService.getJobs({ endDateGt, endDateGte, endDateLt, endDateLte,
executorClass, hostname, isAlive, jobState, jobType, limit, offset, orderBy,
startDateGt, startDateGte, startDateLt, startDateLte }) });
+} = {}) => queryClient.prefetchQuery({ queryKey:
Common.UseJobServiceGetJobsKeyFn({ dagId, endDateGt, endDateGte, endDateLt,
endDateLte, executorClass, hostname, isAlive, jobState, jobType, limit, offset,
orderBy, startDateGt, startDateGte, startDateLt, startDateLte }), queryFn: ()
=> JobService.getJobs({ dagId, endDateGt, endDateGte, endDateLt, endDateLte,
executorClass, hostname, isAlive, jobState, jobType, limit, offset, orderBy,
startDateGt, startDateGte, startDateLt, startDateLte }) });
/**
* Get Plugins
* @param data The data for the request.
diff --git a/airflow-core/src/airflow/ui/openapi-gen/queries/queries.ts
b/airflow-core/src/airflow/ui/openapi-gen/queries/queries.ts
index 3d334e51a96..cc0d92736ca 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/queries.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/queries.ts
@@ -1441,13 +1441,15 @@ export const useImportErrorServiceGetImportErrors =
<TData = Common.ImportErrorS
* @param data.offset
* @param data.orderBy Attributes to order by, multi criteria sort is
supported. Prefix with `-` for descending order. Supported attributes: `id,
dag_id, state, job_type, start_date, end_date, latest_heartbeat,
executor_class, hostname, unixname`
* @param data.jobState
+* @param data.dagId
* @param data.jobType
* @param data.hostname
* @param data.executorClass
* @returns JobCollectionResponse Successful Response
* @throws ApiError
*/
-export const useJobServiceGetJobs = <TData =
Common.JobServiceGetJobsDefaultResponse, TError = unknown, TQueryKey extends
Array<unknown> = unknown[]>({ endDateGt, endDateGte, endDateLt, endDateLte,
executorClass, hostname, isAlive, jobState, jobType, limit, offset, orderBy,
startDateGt, startDateGte, startDateLt, startDateLte }: {
+export const useJobServiceGetJobs = <TData =
Common.JobServiceGetJobsDefaultResponse, TError = unknown, TQueryKey extends
Array<unknown> = unknown[]>({ dagId, endDateGt, endDateGte, endDateLt,
endDateLte, executorClass, hostname, isAlive, jobState, jobType, limit, offset,
orderBy, startDateGt, startDateGte, startDateLt, startDateLte }: {
+ dagId?: string;
endDateGt?: string;
endDateGte?: string;
endDateLt?: string;
@@ -1464,7 +1466,7 @@ export const useJobServiceGetJobs = <TData =
Common.JobServiceGetJobsDefaultResp
startDateGte?: string;
startDateLt?: string;
startDateLte?: string;
-} = {}, queryKey?: TQueryKey, options?: Omit<UseQueryOptions<TData, TError>,
"queryKey" | "queryFn">) => useQuery<TData, TError>({ queryKey:
Common.UseJobServiceGetJobsKeyFn({ endDateGt, endDateGte, endDateLt,
endDateLte, executorClass, hostname, isAlive, jobState, jobType, limit, offset,
orderBy, startDateGt, startDateGte, startDateLt, startDateLte }, queryKey),
queryFn: () => JobService.getJobs({ endDateGt, endDateGte, endDateLt,
endDateLte, executorClass, hostname, isAlive, jobState, [...]
+} = {}, queryKey?: TQueryKey, options?: Omit<UseQueryOptions<TData, TError>,
"queryKey" | "queryFn">) => useQuery<TData, TError>({ queryKey:
Common.UseJobServiceGetJobsKeyFn({ dagId, endDateGt, endDateGte, endDateLt,
endDateLte, executorClass, hostname, isAlive, jobState, jobType, limit, offset,
orderBy, startDateGt, startDateGte, startDateLt, startDateLte }, queryKey),
queryFn: () => JobService.getJobs({ dagId, endDateGt, endDateGte, endDateLt,
endDateLte, executorClass, hostname, isAli [...]
/**
* Get Plugins
* @param data The data for the request.
diff --git a/airflow-core/src/airflow/ui/openapi-gen/queries/suspense.ts
b/airflow-core/src/airflow/ui/openapi-gen/queries/suspense.ts
index a83ac5cb0ae..62139d375db 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/queries/suspense.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/queries/suspense.ts
@@ -1441,13 +1441,15 @@ export const
useImportErrorServiceGetImportErrorsSuspense = <TData = Common.Impo
* @param data.offset
* @param data.orderBy Attributes to order by, multi criteria sort is
supported. Prefix with `-` for descending order. Supported attributes: `id,
dag_id, state, job_type, start_date, end_date, latest_heartbeat,
executor_class, hostname, unixname`
* @param data.jobState
+* @param data.dagId
* @param data.jobType
* @param data.hostname
* @param data.executorClass
* @returns JobCollectionResponse Successful Response
* @throws ApiError
*/
-export const useJobServiceGetJobsSuspense = <TData =
Common.JobServiceGetJobsDefaultResponse, TError = unknown, TQueryKey extends
Array<unknown> = unknown[]>({ endDateGt, endDateGte, endDateLt, endDateLte,
executorClass, hostname, isAlive, jobState, jobType, limit, offset, orderBy,
startDateGt, startDateGte, startDateLt, startDateLte }: {
+export const useJobServiceGetJobsSuspense = <TData =
Common.JobServiceGetJobsDefaultResponse, TError = unknown, TQueryKey extends
Array<unknown> = unknown[]>({ dagId, endDateGt, endDateGte, endDateLt,
endDateLte, executorClass, hostname, isAlive, jobState, jobType, limit, offset,
orderBy, startDateGt, startDateGte, startDateLt, startDateLte }: {
+ dagId?: string;
endDateGt?: string;
endDateGte?: string;
endDateLt?: string;
@@ -1464,7 +1466,7 @@ export const useJobServiceGetJobsSuspense = <TData =
Common.JobServiceGetJobsDef
startDateGte?: string;
startDateLt?: string;
startDateLte?: string;
-} = {}, queryKey?: TQueryKey, options?: Omit<UseQueryOptions<TData, TError>,
"queryKey" | "queryFn">) => useSuspenseQuery<TData, TError>({ queryKey:
Common.UseJobServiceGetJobsKeyFn({ endDateGt, endDateGte, endDateLt,
endDateLte, executorClass, hostname, isAlive, jobState, jobType, limit, offset,
orderBy, startDateGt, startDateGte, startDateLt, startDateLte }, queryKey),
queryFn: () => JobService.getJobs({ endDateGt, endDateGte, endDateLt,
endDateLte, executorClass, hostname, isAlive, jo [...]
+} = {}, queryKey?: TQueryKey, options?: Omit<UseQueryOptions<TData, TError>,
"queryKey" | "queryFn">) => useSuspenseQuery<TData, TError>({ queryKey:
Common.UseJobServiceGetJobsKeyFn({ dagId, endDateGt, endDateGte, endDateLt,
endDateLte, executorClass, hostname, isAlive, jobState, jobType, limit, offset,
orderBy, startDateGt, startDateGte, startDateLt, startDateLte }, queryKey),
queryFn: () => JobService.getJobs({ dagId, endDateGt, endDateGte, endDateLt,
endDateLte, executorClass, hostnam [...]
/**
* Get Plugins
* @param data The data for the request.
diff --git a/airflow-core/src/airflow/ui/openapi-gen/requests/services.gen.ts
b/airflow-core/src/airflow/ui/openapi-gen/requests/services.gen.ts
index 0d225056ac2..ca8d6339321 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/requests/services.gen.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/requests/services.gen.ts
@@ -3430,6 +3430,7 @@ export class JobService {
* @param data.offset
* @param data.orderBy Attributes to order by, multi criteria sort is
supported. Prefix with `-` for descending order. Supported attributes: `id,
dag_id, state, job_type, start_date, end_date, latest_heartbeat,
executor_class, hostname, unixname`
* @param data.jobState
+ * @param data.dagId
* @param data.jobType
* @param data.hostname
* @param data.executorClass
@@ -3454,6 +3455,7 @@ export class JobService {
offset: data.offset,
order_by: data.orderBy,
job_state: data.jobState,
+ dag_id: data.dagId,
job_type: data.jobType,
hostname: data.hostname,
executor_class: data.executorClass
diff --git a/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts
b/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts
index 3bbc72ebc95..235c75ec085 100644
--- a/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts
+++ b/airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts
@@ -4191,6 +4191,7 @@ export type GetImportErrorsData = {
export type GetImportErrorsResponse = ImportErrorCollectionResponse;
export type GetJobsData = {
+ dagId?: string | null;
endDateGt?: string | null;
endDateGte?: string | null;
endDateLt?: string | null;
diff --git
a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_job.py
b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_job.py
index a350585dc3a..8e9538d3f63 100644
--- a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_job.py
+++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_job.py
@@ -170,6 +170,23 @@ class TestGetJobs(TestJobEndpoint):
}
assert resp_job == expected_job
+ def test_get_jobs_filters_by_dag_id(self, test_client, session: Session):
+ clear_db_jobs()
+ session.add_all(
+ [
+ Job(dag_id="target_dag", state=JobState.RUNNING,
job_type="SchedulerJob"),
+ Job(dag_id="other_dag", state=JobState.SUCCESS,
job_type="SchedulerJob"),
+ ]
+ )
+ session.commit()
+
+ response = test_client.get("/jobs", params={"dag_id": "target_dag"})
+
+ assert response.status_code == 200
+ response_json = response.json()
+ assert response_json["total_entries"] == 1
+ assert response_json["jobs"][0]["dag_id"] == "target_dag"
+
def test_should_raises_401_unauthenticated(self,
unauthenticated_test_client):
response = unauthenticated_test_client.get("/jobs")
assert response.status_code == 401
diff --git a/airflow-ctl/src/airflowctl/api/operations.py
b/airflow-ctl/src/airflowctl/api/operations.py
index f58ad78adeb..c682608e85f 100644
--- a/airflow-ctl/src/airflowctl/api/operations.py
+++ b/airflow-ctl/src/airflowctl/api/operations.py
@@ -685,9 +685,27 @@ class JobsOperations(BaseOperations):
job_type: str | None = None,
hostname: str | None = None,
is_alive: bool | None = None,
+ dag_id: str | None = None,
+ state: str | None = None,
+ limit: int | None = None,
+ offset: int | None = None,
+ order_by: str | None = None,
) -> JobCollectionResponse | ServerResponseError:
"""List all jobs."""
- params = _build_query_params(job_type=job_type or None,
hostname=hostname or None, is_alive=is_alive)
+ params = _build_query_params(
+ job_type=job_type or None,
+ hostname=hostname or None,
+ is_alive=is_alive,
+ dag_id=dag_id or None,
+ job_state=state or None,
+ order_by=order_by or "-start_date",
+ limit=limit,
+ offset=offset,
+ )
+
+ if limit is not None or offset is not None:
+ self.response = self.client.get("jobs", params=params)
+ return
JobCollectionResponse.model_validate_json(self.response.content)
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 dc5cdd396db..384e6ecae3d 100644
--- a/airflow-ctl/tests/airflow_ctl/api/test_operations.py
+++ b/airflow-ctl/tests/airflow_ctl/api/test_operations.py
@@ -1359,6 +1359,9 @@ class TestJobsOperations:
assert params["job_type"] == "job_type"
assert params["hostname"] == "hostname"
assert params["is_alive"] == "true"
+ assert params["dag_id"] == "dag_id"
+ assert params["job_state"] == "success"
+ assert params["order_by"] == "-start_date"
assert params["limit"] == "50"
return httpx.Response(200,
json=json.loads(self.job_collection_response.model_dump_json()))
@@ -1367,33 +1370,84 @@ class TestJobsOperations:
job_type="job_type",
hostname="hostname",
is_alive=True,
+ dag_id="dag_id",
+ state="success",
+ order_by="-start_date",
)
assert response == self.job_collection_response
+ def test_list_with_limit_returns_single_page(self):
+ requests = []
+
+ def handle_request(request: httpx.Request) -> httpx.Response:
+ requests.append(request)
+ assert request.url.path == "/api/v2/jobs"
+ params = dict(request.url.params)
+ assert params["limit"] == "1"
+ assert params["order_by"] == "-start_date"
+ return httpx.Response(
+ 200,
+ json={
+
**json.loads(self.job_collection_response.model_dump_json()),
+ "total_entries": 2,
+ },
+ )
+
+ client = make_api_client(transport=httpx.MockTransport(handle_request))
+ response = client.jobs.list(limit=1)
+
+ assert response.jobs == self.job_collection_response.jobs
+ assert response.total_entries == 2
+ assert len(requests) == 1
+
+ def test_list_with_limit_preserves_explicit_order_by(self):
+ def handle_request(request: httpx.Request) -> httpx.Response:
+ assert request.url.path == "/api/v2/jobs"
+ params = dict(request.url.params)
+ assert params["limit"] == "1"
+ assert params["order_by"] == "id"
+ return httpx.Response(200,
json=json.loads(self.job_collection_response.model_dump_json()))
+
+ client = make_api_client(transport=httpx.MockTransport(handle_request))
+ response = client.jobs.list(limit=1, order_by="id")
+
+ assert response == self.job_collection_response
+
@pytest.mark.parametrize(
- ("job_type", "hostname", "is_alive", "expected_subset"),
+ ("job_type", "hostname", "is_alive", "dag_id", "state",
"expected_subset"),
[
- (None, None, None, {}),
- ("scheduler", None, None, {"job_type": "scheduler"}),
- (None, "host-a", None, {"hostname": "host-a"}),
- (None, None, False, {"is_alive": "false"}),
+ (None, None, None, None, None, {}),
+ ("scheduler", None, None, None, None, {"job_type": "scheduler"}),
+ (None, "host-a", None, None, None, {"hostname": "host-a"}),
+ (None, None, False, None, None, {"is_alive": "false"}),
+ (None, None, None, "dag_a", None, {"dag_id": "dag_a"}),
+ (None, None, None, None, "running", {"job_state": "running"}),
],
)
- def test_list_omits_empty_filters(self, job_type, hostname, is_alive,
expected_subset):
+ def test_list_omits_empty_filters(self, job_type, hostname, is_alive,
dag_id, state, expected_subset):
def handle_request(request: httpx.Request) -> httpx.Response:
assert request.url.path == "/api/v2/jobs"
params = dict(request.url.params)
assert params["limit"] == "50"
+ assert params["order_by"] == "-start_date"
for key, value in expected_subset.items():
assert params[key] == value
assert ("job_type" in params) is ("job_type" in expected_subset)
assert ("hostname" in params) is ("hostname" in expected_subset)
assert ("is_alive" in params) is ("is_alive" in expected_subset)
+ assert ("dag_id" in params) is ("dag_id" in expected_subset)
+ assert ("job_state" in params) is ("job_state" in expected_subset)
return httpx.Response(200,
json=json.loads(self.job_collection_response.model_dump_json()))
client = make_api_client(transport=httpx.MockTransport(handle_request))
- response = client.jobs.list(job_type=job_type, hostname=hostname,
is_alive=is_alive)
+ response = client.jobs.list(
+ job_type=job_type,
+ hostname=hostname,
+ is_alive=is_alive,
+ dag_id=dag_id,
+ state=state,
+ )
assert response == self.job_collection_response
