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 8173fd5fe9c61..0354bfa1ae2c1 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 @@ -5305,6 +5305,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 b56c78f2de659..3e7e16bb69ad0 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 a8f08d3df78c0..bb22a98d1dba3 100644 --- a/airflow-core/src/airflow/ui/openapi-gen/queries/common.ts +++ b/airflow-core/src/airflow/ui/openapi-gen/queries/common.ts @@ -692,7 +692,8 @@ export const UseImportErrorServiceGetImportErrorsKeyFn = ({ bundleName, filename export type JobServiceGetJobsDefaultResponse = Awaited>; export type JobServiceGetJobsQueryResult = UseQueryResult; 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; @@ -709,7 +710,7 @@ export const UseJobServiceGetJobsKeyFn = ({ endDateGt, endDateGte, endDateLt, en startDateGte?: string; startDateLt?: string; startDateLte?: string; -} = {}, queryKey?: Array) => [useJobServiceGetJobsKey, ...(queryKey ?? [{ endDateGt, endDateGte, endDateLt, endDateLte, executorClass, hostname, isAlive, jobState, jobType, limit, offset, orderBy, startDateGt, startDateGte, startDateLt, startDateLte }])]; +} = {}, queryKey?: Array) => [useJobServiceGetJobsKey, ...(queryKey ?? [{ dagId, endDateGt, endDateGte, endDateLt, endDateLte, executorClass, hostname, isAlive, jobState, jobType, limit, offset, orderBy, startDateGt, startDateGte, startDateLt, startDateLte }])]; export type PluginServiceGetPluginsDefaultResponse = Awaited>; export type PluginServiceGetPluginsQueryResult = UseQueryResult; 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 34eff20314070..33921b93ada1d 100644 --- a/airflow-core/src/airflow/ui/openapi-gen/queries/ensureQueryData.ts +++ b/airflow-core/src/airflow/ui/openapi-gen/queries/ensureQueryData.ts @@ -1420,13 +1420,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; @@ -1443,7 +1445,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 170a17f9e58ae..cc64fdfd49949 100644 --- a/airflow-core/src/airflow/ui/openapi-gen/queries/prefetch.ts +++ b/airflow-core/src/airflow/ui/openapi-gen/queries/prefetch.ts @@ -1420,13 +1420,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; @@ -1443,7 +1445,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 7efadc2e4f8fa..8d7955bd07fdc 100644 --- a/airflow-core/src/airflow/ui/openapi-gen/queries/queries.ts +++ b/airflow-core/src/airflow/ui/openapi-gen/queries/queries.ts @@ -1420,13 +1420,15 @@ export const useImportErrorServiceGetImportErrors = = unknown[]>({ endDateGt, endDateGte, endDateLt, endDateLte, executorClass, hostname, isAlive, jobState, jobType, limit, offset, orderBy, startDateGt, startDateGte, startDateLt, startDateLte }: { +export const useJobServiceGetJobs = = 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; @@ -1443,7 +1445,7 @@ export const useJobServiceGetJobs = , "queryKey" | "queryFn">) => useQuery({ 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, jobType, limit, offset, orderBy, startDateGt, startDateGte, startDateLt, startDateLte }) as TData, ...options }); +} = {}, queryKey?: TQueryKey, options?: Omit, "queryKey" | "queryFn">) => useQuery({ 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, isAlive, jobState, jobType, limit, offset, orderBy, startDateGt, startDateGte, startDateLt, startDateLte }) as TData, ...options }); /** * 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 d4e694554d0ff..0b9fb7b59987a 100644 --- a/airflow-core/src/airflow/ui/openapi-gen/queries/suspense.ts +++ b/airflow-core/src/airflow/ui/openapi-gen/queries/suspense.ts @@ -1420,13 +1420,15 @@ export const useImportErrorServiceGetImportErrorsSuspense = = unknown[]>({ endDateGt, endDateGte, endDateLt, endDateLte, executorClass, hostname, isAlive, jobState, jobType, limit, offset, orderBy, startDateGt, startDateGte, startDateLt, startDateLte }: { +export const useJobServiceGetJobsSuspense = = 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; @@ -1443,7 +1445,7 @@ export const useJobServiceGetJobsSuspense = , "queryKey" | "queryFn">) => useSuspenseQuery({ 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, jobType, limit, offset, orderBy, startDateGt, startDateGte, startDateLt, startDateLte }) as TData, ...options }); +} = {}, queryKey?: TQueryKey, options?: Omit, "queryKey" | "queryFn">) => useSuspenseQuery({ 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, isAlive, jobState, jobType, limit, offset, orderBy, startDateGt, startDateGte, startDateLt, startDateLte }) as TData, ...options }); /** * 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 0356a4a364197..2bd607bd14d8e 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 @@ -3394,6 +3394,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 @@ -3418,6 +3419,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 acec153fd1fe2..1dd79fc66ba6b 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 @@ -4145,6 +4145,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 a350585dc3aaf..8e9538d3f634d 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 @@ def test_get_jobs( } 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 f58ad78adeb37..c682608e85fd3 100644 --- a/airflow-ctl/src/airflowctl/api/operations.py +++ b/airflow-ctl/src/airflowctl/api/operations.py @@ -685,9 +685,27 @@ def list( 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 dc5cdd396db8b..384e6ecae3dd5 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 @@ def handle_request(request: httpx.Request) -> httpx.Response: 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 @@ def handle_request(request: httpx.Request) -> httpx.Response: 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