You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "jason810496 (via GitHub)" <gi...@apache.org> on 2024/11/10 09:37:57 UTC
[PR] AIP-81 Add Lists Jobs with filters API [airflow]
jason810496 opened a new pull request, #43859:
URL: https://github.com/apache/airflow/pull/43859
Closes: #43660
Related: #43657
### Feature
I’ve placed the endpoint in `airflow/api_fastapi/core_api/routes/public/job.py`, but I’m not certain if this is the most appropriate location—perhaps `airflow/api_fastapi/core_api/routes/cli/job.py` would also fine ?
The query parameters, response data models, and implementation are consistent with the rest of the public API, utilizing `common.parameters` and `paginated_select`.
### Tests
The test cases are refactored from the original [airflow/tests/cli/commands/test_jobs_command.py](https://github.com/apache/airflow/blob/main/tests/cli/commands/test_jobs_command.py), as the CLI should yield the same results when leveraging the API instead of direct DB access.
### Question
Hi @bugraoz93, I’m curious about the distinction between this PR’s endpoint (List Jobs with filters) and #43661 (List Jobs), as both endpoints seem to provide job listings. From my perspective, calling this "List Jobs with filters" endpoint without query parameters would yield the same response as the "List Jobs" endpoint.
cc @josix
--
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] AIP-81 Add Lists Jobs with Filters API [airflow]
Posted by "bugraoz93 (via GitHub)" <gi...@apache.org>.
bugraoz93 commented on PR #43859:
URL: https://github.com/apache/airflow/pull/43859#issuecomment-2466811425
>I’ve placed the endpoint in airflow/api_fastapi/core_api/routes/public/job.py, but I’m not certain if this is the most appropriate location—perhaps airflow/api_fastapi/core_api/routes/cli/job.py would also fine ?
The query parameters, response data models, and implementation are consistent with the rest of the public API, utilizing common.parameters and paginated_select.
@jason810496 I missed including this part in my previous answer. I responded from my mind but forgot to put it into text. :)
I think it makes more sense to keep the endpoint under `Core API`. Since most `CLI` calls will still rely on `Core API`, creating a separate category for just a few endpoints could add unnecessary complexity. By sticking to `Core API`, we stay consistent with the rest of the API and follow the same rules across the board. This way, we avoid extra maintenance overhead and keep things simple as we scale.
--
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] AIP-84 Add Lists Jobs with Filters API [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on PR #43859:
URL: https://github.com/apache/airflow/pull/43859#issuecomment-2493836512
Most likely the CI image used to run the pre-commit hook (`in_container` script) still has an old pydantic version.
You can run:
```shell
breeze ci-image build --python 3.9
```
To refresh the images. And you should get the appropriate generated spec.
--
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] AIP-84 Add Lists Jobs with Filters API [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #43859:
URL: https://github.com/apache/airflow/pull/43859#discussion_r1837921946
##########
airflow/api_fastapi/core_api/routes/public/job.py:
##########
@@ -0,0 +1,128 @@
+# 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 fastapi import Depends, status
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+from typing_extensions import Annotated
+
+from airflow.api_fastapi.common.db.common import (
+ get_session,
+ paginated_select,
+)
+from airflow.api_fastapi.common.parameters import (
+ QueryLimit,
+ QueryOffset,
+ RangeFilter,
+ SortParam,
+ datetime_range_filter_factory,
+)
+from airflow.api_fastapi.common.router import AirflowRouter
+from airflow.api_fastapi.core_api.datamodels.job import (
+ JobCollectionResponse,
+ JobResponse,
+)
+from airflow.api_fastapi.core_api.openapi.exceptions import create_openapi_http_exception_doc
+from airflow.jobs.job import Job
+from airflow.utils.state import JobState
+
+job_router = AirflowRouter(tags=["Job"], prefix="/jobs")
+
+
+@job_router.get(
+ "/",
+ responses=create_openapi_http_exception_doc(
+ [status.HTTP_400_BAD_REQUEST, status.HTTP_401_UNAUTHORIZED, status.HTTP_403_FORBIDDEN]
+ ),
+)
+def get_jobs(
+ start_date_range: Annotated[
+ RangeFilter,
+ Depends(datetime_range_filter_factory("start_date", Job)),
+ ],
+ end_date_range: Annotated[
+ RangeFilter,
+ Depends(datetime_range_filter_factory("end_date", Job)),
+ ],
+ limit: QueryLimit,
+ offset: QueryOffset,
+ order_by: Annotated[
+ SortParam,
+ Depends(
+ SortParam(
+ [
+ "id",
+ "dag_id",
+ "state",
+ "job_type",
+ "start_date",
+ "end_date",
+ "latest_heartbeat",
+ "executor_class",
+ "hostname",
+ "unixname",
+ ],
+ Job,
+ ).dynamic_depends(default="id")
+ ),
+ ],
+ session: Annotated[Session, Depends(get_session)],
+ state: str | None = None,
+ job_type: str | None = None,
+ hostname: str | None = None,
+ executor_class: str | None = None,
+ is_alive: bool | None = None,
+) -> JobCollectionResponse:
+ """Get all jobs."""
+ base_select = select(Job).where(Job.state == JobState.RUNNING).order_by(Job.latest_heartbeat.desc())
+ # TODO: Refactor using the `FilterParam` class in commit `574b72e41cc5ed175a2bbf4356522589b836bb11`
+ if state:
+ base_select = base_select.where(Job.state == state)
+ if job_type:
+ base_select = base_select.where(Job.job_type == job_type)
+ if hostname:
+ base_select = base_select.where(Job.hostname == hostname)
+ if executor_class:
+ base_select = base_select.where(Job.executor_class == executor_class)
Review Comment:
Hi @pierrejeambrun,
I'm currently working on `FilterParam` as we discussed in https://github.com/apache/airflow/pull/43407#pullrequestreview-2405200579. The intended usage would be something like `Annotated[ParamFilter[int], Depends(filter_param_factory(filter_name))]`, and the implementation will resemble the structure here: https://github.com/apache/airflow/blob/574b72e41cc5ed175a2bbf4356522589b836bb11/airflow/api_fastapi/common/parameters.py#L222-L268 but will incorporate `FilterOptionEnum`. I believe this could also be refactored within the current WIP.
--
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] AIP-84 Add Lists Jobs with Filters API [airflow]
Posted by "rawwar (via GitHub)" <gi...@apache.org>.
rawwar commented on code in PR #43859:
URL: https://github.com/apache/airflow/pull/43859#discussion_r1846370505
##########
airflow/api_fastapi/core_api/routes/public/job.py:
##########
@@ -0,0 +1,128 @@
+# 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 fastapi import Depends, status
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+from typing_extensions import Annotated
+
+from airflow.api_fastapi.common.db.common import (
+ get_session,
+ paginated_select,
+)
+from airflow.api_fastapi.common.parameters import (
+ QueryJobExecutorClassFilter,
+ QueryJobHostnameFilter,
+ QueryJobStateFilter,
+ QueryJobTypeFilter,
+ QueryLimit,
+ QueryOffset,
+ RangeFilter,
+ SortParam,
+ datetime_range_filter_factory,
+)
+from airflow.api_fastapi.common.router import AirflowRouter
+from airflow.api_fastapi.core_api.datamodels.job import (
+ JobCollectionResponse,
+ JobResponse,
+)
+from airflow.api_fastapi.core_api.openapi.exceptions import create_openapi_http_exception_doc
+from airflow.jobs.job import Job
+from airflow.utils.state import JobState
+
+job_router = AirflowRouter(tags=["Job"], prefix="/jobs")
+
+
+@job_router.get(
+ "/",
+ responses=create_openapi_http_exception_doc(
+ [status.HTTP_400_BAD_REQUEST, status.HTTP_401_UNAUTHORIZED, status.HTTP_403_FORBIDDEN]
Review Comment:
```suggestion
[status.HTTP_400_BAD_REQUEST]
```
--
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] AIP-81 Add Lists Jobs with Filters API [airflow]
Posted by "bugraoz93 (via GitHub)" <gi...@apache.org>.
bugraoz93 commented on PR #43859:
URL: https://github.com/apache/airflow/pull/43859#issuecomment-2466738962
Hey @jason810496,
That’s a good question, and I’m glad you raised it! @rawwar and I had a similar discussion, and it seems we all have similar concerns.
At first, I thought keeping `"List Jobs"` and `"List Jobs with Filters"` as separate endpoints made sense for clarity. But after thinking it through more and taking a closer look, I believe combining them into a single "List Jobs" endpoint with optional query parameters for filters might be a better approach. This way, the endpoint would return all jobs by default, with filtering options (as a `QueryParameter`) as needed. It would make it both straightforward and easy to maintain.
Thanks for raising this! I am happy to keep the discussion going if others have thoughts on this approach. Otherwise, let’s go ahead and merge it into one endpoint.
cc: @josix @rawwar
--
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] AIP-84 Add Lists Jobs with Filters API [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #43859:
URL: https://github.com/apache/airflow/pull/43859#discussion_r1838201230
##########
tests/api_fastapi/core_api/routes/public/test_job.py:
##########
@@ -0,0 +1,166 @@
+# 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 typing import Literal
+
+import pytest
+
+from airflow.jobs.job import Job
+from airflow.jobs.scheduler_job_runner import SchedulerJobRunner
+from airflow.utils.session import provide_session
+from airflow.utils.state import JobState, State
+
+from tests_common.test_utils.db import clear_db_jobs
+
+pytestmark = pytest.mark.db_test
+
+TESTCASE_TYPE = Literal[
+ "should_report_success_for_one_working_scheduler",
+ "should_report_success_for_one_working_scheduler_with_hostname",
+ "should_report_success_for_ha_schedulers",
+ "should_ignore_not_running_jobs",
+ "should_raise_exception_for_multiple_scheduler_on_one_host",
+]
+TESTCASE_ONE_SCHEDULER = "should_report_success_for_one_working_scheduler"
+TESTCASE_ONE_SCHEDULER_WITH_HOSTNAME = "should_report_success_for_one_working_scheduler_with_hostname"
+TESTCASE_HA_SCHEDULERS = "should_report_success_for_ha_schedulers"
+TESTCASE_IGNORE_NOT_RUNNING = "should_ignore_not_running_jobs"
+TESTCASE_MULTIPLE_SCHEDULERS_ON_ONE_HOST = "should_raise_exception_for_multiple_scheduler_on_one_host"
+TESTCASE_MULTIPLE_RUNNER = "should_report_success_for_multiple_runners"
+
+
+class TestJobEndpoint:
+ """Common class for /public/jobs related unit tests."""
+
+ scheduler_jobs: list[Job] | None = None
+ job_runners: list[SchedulerJobRunner] | None = None
+
+ def _setup_should_report_success_for_one_working_scheduler(self, session=None):
+ scheduler_job = Job()
+ job_runner = SchedulerJobRunner(job=scheduler_job)
+ scheduler_job.state = State.RUNNING
+ session.add(scheduler_job)
+ session.commit()
+ self.scheduler_jobs.append(scheduler_job)
+ self.job_runners.append(job_runner)
+ scheduler_job.heartbeat(heartbeat_callback=job_runner.heartbeat_callback)
+
+ def _setup_should_report_success_for_one_working_scheduler_with_hostname(self, session=None):
+ scheduler_job = Job()
+ job_runner = SchedulerJobRunner(job=scheduler_job)
+ scheduler_job.state = State.RUNNING
+ scheduler_job.hostname = "HOSTNAME"
+ session.add(scheduler_job)
+ self.scheduler_jobs.append(scheduler_job)
+ self.job_runners.append(job_runner)
+ session.commit()
+ scheduler_job.heartbeat(heartbeat_callback=job_runner.heartbeat_callback)
+
+ def _setup_should_report_success_for_ha_schedulers(self, session=None):
+ for _ in range(3):
+ scheduler_job = Job()
+ job_runner = SchedulerJobRunner(job=scheduler_job)
+ scheduler_job.state = State.RUNNING
+ session.add(scheduler_job)
+ self.scheduler_jobs.append(scheduler_job)
+ self.job_runners.append(job_runner)
+ session.commit()
+ scheduler_job.heartbeat(heartbeat_callback=job_runner.heartbeat_callback)
+
+ def _setup_should_ignore_not_running_jobs(self, session=None):
+ for _ in range(3):
+ scheduler_job = Job()
+ job_runner = SchedulerJobRunner(job=scheduler_job)
+ scheduler_job.state = JobState.FAILED
+ session.add(scheduler_job)
+ self.scheduler_jobs.append(scheduler_job)
+ self.job_runners.append(job_runner)
+ session.commit()
+
+ def _setup_should_raise_exception_for_multiple_scheduler_on_one_host(self, session=None):
+ for _ in range(3):
+ scheduler_job = Job()
+ job_runner = SchedulerJobRunner(job=scheduler_job)
+ job_runner.job = scheduler_job
+ scheduler_job.state = State.RUNNING
+ scheduler_job.hostname = "HOSTNAME"
+ session.add(scheduler_job)
+ self.scheduler_jobs.append(scheduler_job)
+ self.job_runners.append(job_runner)
+ session.commit()
+ scheduler_job.heartbeat(heartbeat_callback=job_runner.heartbeat_callback)
+
+ @provide_session
+ def setup(self, testcase: TESTCASE_TYPE, session=None) -> None:
+ """
+ Setup testcase at runtime based on the `testcase` provided by `pytest.mark.parametrize`.
+ """
+ clear_db_jobs()
+ self.scheduler_jobs = []
+ self.job_runners = []
+ setup_method = getattr(self, f"_setup_{testcase}")
+ setup_method(session)
+
+ def teardown_method(self) -> None:
+ if self.job_runners:
+ for job_runner in self.job_runners:
+ if job_runner.processor_agent:
+ job_runner.processor_agent.end()
+ clear_db_jobs()
+
+
+class TestGetJobs(TestJobEndpoint):
+ @pytest.mark.parametrize(
+ "testcase, query_params, expected_status_code, expected_total_entries",
+ [
+ # original testcases refactor from tests/cli/commands/test_jobs_command.py
+ (TESTCASE_ONE_SCHEDULER, {}, 200, 1),
+ (TESTCASE_ONE_SCHEDULER_WITH_HOSTNAME, {"hostname": "HOSTNAME"}, 200, 1),
+ (TESTCASE_HA_SCHEDULERS, {"limit": 100}, 200, 3),
+ (TESTCASE_IGNORE_NOT_RUNNING, {}, 200, 0),
+ (TESTCASE_MULTIPLE_SCHEDULERS_ON_ONE_HOST, {"limit": 100}, 200, 3),
+ ],
+ )
+ def test_get_jobs(
+ self, test_client, testcase, query_params, expected_status_code, expected_total_entries
+ ):
+ # setup testcase at runtime based on the `testcase` parameter
+ self.setup(testcase)
+ response = test_client.get("/public/jobs/", params=query_params)
+ assert response.status_code == expected_status_code
+ if expected_status_code != 200:
+ return
+ response_json = response.json()
+ assert response_json["total_entries"] == expected_total_entries
+
+ for idx, resp_job in enumerate(response_json["jobs"]):
+ expected_job = {
+ "id": self.scheduler_jobs[idx].id,
+ "dag_id": None,
+ "state": "running",
+ "job_type": "SchedulerJob",
+ "start_date": self.scheduler_jobs[idx].start_date.isoformat().replace("+00:00", "Z"),
Review Comment:
I don't think this stringformat exist yet. You need to create it and put it in the a test shared file.
--
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] AIP-84 Add Lists Jobs with Filters API [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #43859:
URL: https://github.com/apache/airflow/pull/43859#discussion_r1838201230
##########
tests/api_fastapi/core_api/routes/public/test_job.py:
##########
@@ -0,0 +1,166 @@
+# 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 typing import Literal
+
+import pytest
+
+from airflow.jobs.job import Job
+from airflow.jobs.scheduler_job_runner import SchedulerJobRunner
+from airflow.utils.session import provide_session
+from airflow.utils.state import JobState, State
+
+from tests_common.test_utils.db import clear_db_jobs
+
+pytestmark = pytest.mark.db_test
+
+TESTCASE_TYPE = Literal[
+ "should_report_success_for_one_working_scheduler",
+ "should_report_success_for_one_working_scheduler_with_hostname",
+ "should_report_success_for_ha_schedulers",
+ "should_ignore_not_running_jobs",
+ "should_raise_exception_for_multiple_scheduler_on_one_host",
+]
+TESTCASE_ONE_SCHEDULER = "should_report_success_for_one_working_scheduler"
+TESTCASE_ONE_SCHEDULER_WITH_HOSTNAME = "should_report_success_for_one_working_scheduler_with_hostname"
+TESTCASE_HA_SCHEDULERS = "should_report_success_for_ha_schedulers"
+TESTCASE_IGNORE_NOT_RUNNING = "should_ignore_not_running_jobs"
+TESTCASE_MULTIPLE_SCHEDULERS_ON_ONE_HOST = "should_raise_exception_for_multiple_scheduler_on_one_host"
+TESTCASE_MULTIPLE_RUNNER = "should_report_success_for_multiple_runners"
+
+
+class TestJobEndpoint:
+ """Common class for /public/jobs related unit tests."""
+
+ scheduler_jobs: list[Job] | None = None
+ job_runners: list[SchedulerJobRunner] | None = None
+
+ def _setup_should_report_success_for_one_working_scheduler(self, session=None):
+ scheduler_job = Job()
+ job_runner = SchedulerJobRunner(job=scheduler_job)
+ scheduler_job.state = State.RUNNING
+ session.add(scheduler_job)
+ session.commit()
+ self.scheduler_jobs.append(scheduler_job)
+ self.job_runners.append(job_runner)
+ scheduler_job.heartbeat(heartbeat_callback=job_runner.heartbeat_callback)
+
+ def _setup_should_report_success_for_one_working_scheduler_with_hostname(self, session=None):
+ scheduler_job = Job()
+ job_runner = SchedulerJobRunner(job=scheduler_job)
+ scheduler_job.state = State.RUNNING
+ scheduler_job.hostname = "HOSTNAME"
+ session.add(scheduler_job)
+ self.scheduler_jobs.append(scheduler_job)
+ self.job_runners.append(job_runner)
+ session.commit()
+ scheduler_job.heartbeat(heartbeat_callback=job_runner.heartbeat_callback)
+
+ def _setup_should_report_success_for_ha_schedulers(self, session=None):
+ for _ in range(3):
+ scheduler_job = Job()
+ job_runner = SchedulerJobRunner(job=scheduler_job)
+ scheduler_job.state = State.RUNNING
+ session.add(scheduler_job)
+ self.scheduler_jobs.append(scheduler_job)
+ self.job_runners.append(job_runner)
+ session.commit()
+ scheduler_job.heartbeat(heartbeat_callback=job_runner.heartbeat_callback)
+
+ def _setup_should_ignore_not_running_jobs(self, session=None):
+ for _ in range(3):
+ scheduler_job = Job()
+ job_runner = SchedulerJobRunner(job=scheduler_job)
+ scheduler_job.state = JobState.FAILED
+ session.add(scheduler_job)
+ self.scheduler_jobs.append(scheduler_job)
+ self.job_runners.append(job_runner)
+ session.commit()
+
+ def _setup_should_raise_exception_for_multiple_scheduler_on_one_host(self, session=None):
+ for _ in range(3):
+ scheduler_job = Job()
+ job_runner = SchedulerJobRunner(job=scheduler_job)
+ job_runner.job = scheduler_job
+ scheduler_job.state = State.RUNNING
+ scheduler_job.hostname = "HOSTNAME"
+ session.add(scheduler_job)
+ self.scheduler_jobs.append(scheduler_job)
+ self.job_runners.append(job_runner)
+ session.commit()
+ scheduler_job.heartbeat(heartbeat_callback=job_runner.heartbeat_callback)
+
+ @provide_session
+ def setup(self, testcase: TESTCASE_TYPE, session=None) -> None:
+ """
+ Setup testcase at runtime based on the `testcase` provided by `pytest.mark.parametrize`.
+ """
+ clear_db_jobs()
+ self.scheduler_jobs = []
+ self.job_runners = []
+ setup_method = getattr(self, f"_setup_{testcase}")
+ setup_method(session)
+
+ def teardown_method(self) -> None:
+ if self.job_runners:
+ for job_runner in self.job_runners:
+ if job_runner.processor_agent:
+ job_runner.processor_agent.end()
+ clear_db_jobs()
+
+
+class TestGetJobs(TestJobEndpoint):
+ @pytest.mark.parametrize(
+ "testcase, query_params, expected_status_code, expected_total_entries",
+ [
+ # original testcases refactor from tests/cli/commands/test_jobs_command.py
+ (TESTCASE_ONE_SCHEDULER, {}, 200, 1),
+ (TESTCASE_ONE_SCHEDULER_WITH_HOSTNAME, {"hostname": "HOSTNAME"}, 200, 1),
+ (TESTCASE_HA_SCHEDULERS, {"limit": 100}, 200, 3),
+ (TESTCASE_IGNORE_NOT_RUNNING, {}, 200, 0),
+ (TESTCASE_MULTIPLE_SCHEDULERS_ON_ONE_HOST, {"limit": 100}, 200, 3),
+ ],
+ )
+ def test_get_jobs(
+ self, test_client, testcase, query_params, expected_status_code, expected_total_entries
+ ):
+ # setup testcase at runtime based on the `testcase` parameter
+ self.setup(testcase)
+ response = test_client.get("/public/jobs/", params=query_params)
+ assert response.status_code == expected_status_code
+ if expected_status_code != 200:
+ return
+ response_json = response.json()
+ assert response_json["total_entries"] == expected_total_entries
+
+ for idx, resp_job in enumerate(response_json["jobs"]):
+ expected_job = {
+ "id": self.scheduler_jobs[idx].id,
+ "dag_id": None,
+ "state": "running",
+ "job_type": "SchedulerJob",
+ "start_date": self.scheduler_jobs[idx].start_date.isoformat().replace("+00:00", "Z"),
Review Comment:
I don't think this stringformat exist yet. You need to create it and put it in a test shared file.
--
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] AIP-84 Add Lists Jobs with Filters API [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #43859:
URL: https://github.com/apache/airflow/pull/43859#discussion_r1846550758
##########
airflow/api_fastapi/core_api/routes/public/job.py:
##########
@@ -0,0 +1,128 @@
+# 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 fastapi import Depends, status
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+from typing_extensions import Annotated
+
+from airflow.api_fastapi.common.db.common import (
+ get_session,
+ paginated_select,
+)
+from airflow.api_fastapi.common.parameters import (
+ QueryJobExecutorClassFilter,
+ QueryJobHostnameFilter,
+ QueryJobStateFilter,
+ QueryJobTypeFilter,
+ QueryLimit,
+ QueryOffset,
+ RangeFilter,
+ SortParam,
+ datetime_range_filter_factory,
+)
+from airflow.api_fastapi.common.router import AirflowRouter
+from airflow.api_fastapi.core_api.datamodels.job import (
+ JobCollectionResponse,
+ JobResponse,
+)
+from airflow.api_fastapi.core_api.openapi.exceptions import create_openapi_http_exception_doc
+from airflow.jobs.job import Job
+from airflow.utils.state import JobState
+
+job_router = AirflowRouter(tags=["Job"], prefix="/jobs")
+
+
+@job_router.get(
+ "/",
+ responses=create_openapi_http_exception_doc(
+ [status.HTTP_400_BAD_REQUEST, status.HTTP_401_UNAUTHORIZED, status.HTTP_403_FORBIDDEN]
Review Comment:
Thanks for reminder 🙌 Just resolve.
--
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] AIP-84 Add Lists Jobs with Filters API [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on PR #43859:
URL: https://github.com/apache/airflow/pull/43859#issuecomment-2498061569
One unrelated provider failure, merging:
```
from google.cloud.translate_v3.types import automl_translation
E ImportError: cannot import name 'automl_translation' from 'google.cloud.translate_v3.types' (/usr/local/lib/python3.9/site-packages/google/cloud/translate_v3/types/__init__.py)
----- generated xml file: /files/test_result-providers_google-postgres.xml -----
```
--
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] AIP-84 Add Lists Jobs with Filters API [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on PR #43859:
URL: https://github.com/apache/airflow/pull/43859#issuecomment-2493093935
Just fix the trailing endpoint naming.
I will work on follow up PR to refactor the test case with manual replace to `datetime_zulu_format` utility function 👍
--
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] AIP-84 Add Lists Jobs with Filters API [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #43859:
URL: https://github.com/apache/airflow/pull/43859#discussion_r1850249264
##########
tests_common/test_utils/format_datetime.py:
##########
@@ -0,0 +1,29 @@
+# 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
+
+
+def datetime_zulu_format(dt: datetime) -> str:
+ """Format a datetime object to a string in Zulu time."""
+ return dt.strftime("%Y-%m-%dT%H:%M:%S.%fZ")
+
+
+def datetime_zulu_format_without_ms(dt: datetime) -> str:
+ """Format a datetime object to a string in Zulu time without milliseconds."""
+ return dt.strftime("%Y-%m-%dT%H:%M:%SZ")
Review Comment:
Sorry for overlooking the test failure result, just fix the `datetime` format for `TestGetEventLog`.
--
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] AIP-84 Add Lists Jobs with Filters API [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on PR #43859:
URL: https://github.com/apache/airflow/pull/43859#issuecomment-2488570257
I think rebasing went wrong on the genrated file part.
When you have any conflicts on those (openapi spec, or front end generated code). You can just re-run manually the hooks to refresh the files:
```
pre-commit run generate-openapi-spec --all-files
pre-commit run ts-compile-format-lint-ui --all-files
```
--
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] AIP-84 Add Lists Jobs with Filters API [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #43859:
URL: https://github.com/apache/airflow/pull/43859#discussion_r1852398946
##########
tests_common/test_utils/format_datetime.py:
##########
@@ -0,0 +1,29 @@
+# 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
+
+
+def datetime_zulu_format(dt: datetime) -> str:
+ """Format a datetime object to a string in Zulu time."""
+ return dt.strftime("%Y-%m-%dT%H:%M:%S.%fZ")
+
+
+def datetime_zulu_format_without_ms(dt: datetime) -> str:
+ """Format a datetime object to a string in Zulu time without milliseconds."""
+ return dt.strftime("%Y-%m-%dT%H:%M:%SZ")
Review Comment:
If you want you can also follow up PR to bring that change in other tests. (We are still doing a manual replace)
##########
airflow/api_fastapi/core_api/routes/public/job.py:
##########
@@ -0,0 +1,126 @@
+# 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 fastapi import Depends, status
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+from typing_extensions import Annotated
+
+from airflow.api_fastapi.common.db.common import (
+ get_session,
+ paginated_select,
+)
+from airflow.api_fastapi.common.parameters import (
+ QueryJobExecutorClassFilter,
+ QueryJobHostnameFilter,
+ QueryJobStateFilter,
+ QueryJobTypeFilter,
+ QueryLimit,
+ QueryOffset,
+ RangeFilter,
+ SortParam,
+ datetime_range_filter_factory,
+)
+from airflow.api_fastapi.common.router import AirflowRouter
+from airflow.api_fastapi.core_api.datamodels.job import (
+ JobCollectionResponse,
+ JobResponse,
+)
+from airflow.api_fastapi.core_api.openapi.exceptions import create_openapi_http_exception_doc
+from airflow.jobs.job import Job
+from airflow.utils.state import JobState
+
+job_router = AirflowRouter(tags=["Job"], prefix="/jobs")
+
+
+@job_router.get(
+ "/",
Review Comment:
```suggestion
"",
```
For consistency with other endpoints. Otherwise a trailing `/` will be introduced in the route. (That's a recent change we did in all endpoints to comply with the legacy API).
--
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] AIP-84 Add Lists Jobs with Filters API [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on PR #43859:
URL: https://github.com/apache/airflow/pull/43859#issuecomment-2493819905
This is due to the recent pydantic release that breaks some stuff. Main lower bound has been added and should be better, you most likely not have the appropriate pydantic version locally. Now it should be `>= 2.10.1`(#44284)
--
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] AIP-84 Add Lists Jobs with Filters API [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #43859:
URL: https://github.com/apache/airflow/pull/43859#discussion_r1846622697
##########
tests_common/test_utils/format_datetime.py:
##########
@@ -0,0 +1,29 @@
+# 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
+
+
+def datetime_zulu_format(dt: datetime) -> str:
+ """Format a datetime object to a string in Zulu time."""
+ return dt.strftime("%Y-%m-%dT%H:%M:%S.%fZ")
+
+
+def datetime_zulu_format_without_ms(dt: datetime) -> str:
+ """Format a datetime object to a string in Zulu time without milliseconds."""
+ return dt.strftime("%Y-%m-%dT%H:%M:%SZ")
Review Comment:
nice
--
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] AIP-84 Add Lists Jobs with Filters API [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #43859:
URL: https://github.com/apache/airflow/pull/43859#discussion_r1837787114
##########
tests/api_fastapi/core_api/routes/public/test_job.py:
##########
@@ -0,0 +1,166 @@
+# 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 typing import Literal
+
+import pytest
+
+from airflow.jobs.job import Job
+from airflow.jobs.scheduler_job_runner import SchedulerJobRunner
+from airflow.utils.session import provide_session
+from airflow.utils.state import JobState, State
+
+from tests_common.test_utils.db import clear_db_jobs
+
+pytestmark = pytest.mark.db_test
+
+TESTCASE_TYPE = Literal[
+ "should_report_success_for_one_working_scheduler",
+ "should_report_success_for_one_working_scheduler_with_hostname",
+ "should_report_success_for_ha_schedulers",
+ "should_ignore_not_running_jobs",
+ "should_raise_exception_for_multiple_scheduler_on_one_host",
+]
+TESTCASE_ONE_SCHEDULER = "should_report_success_for_one_working_scheduler"
+TESTCASE_ONE_SCHEDULER_WITH_HOSTNAME = "should_report_success_for_one_working_scheduler_with_hostname"
+TESTCASE_HA_SCHEDULERS = "should_report_success_for_ha_schedulers"
+TESTCASE_IGNORE_NOT_RUNNING = "should_ignore_not_running_jobs"
+TESTCASE_MULTIPLE_SCHEDULERS_ON_ONE_HOST = "should_raise_exception_for_multiple_scheduler_on_one_host"
+TESTCASE_MULTIPLE_RUNNER = "should_report_success_for_multiple_runners"
+
+
+class TestJobEndpoint:
+ """Common class for /public/jobs related unit tests."""
+
+ scheduler_jobs: list[Job] | None = None
+ job_runners: list[SchedulerJobRunner] | None = None
+
+ def _setup_should_report_success_for_one_working_scheduler(self, session=None):
+ scheduler_job = Job()
+ job_runner = SchedulerJobRunner(job=scheduler_job)
+ scheduler_job.state = State.RUNNING
+ session.add(scheduler_job)
+ session.commit()
+ self.scheduler_jobs.append(scheduler_job)
+ self.job_runners.append(job_runner)
+ scheduler_job.heartbeat(heartbeat_callback=job_runner.heartbeat_callback)
+
+ def _setup_should_report_success_for_one_working_scheduler_with_hostname(self, session=None):
+ scheduler_job = Job()
+ job_runner = SchedulerJobRunner(job=scheduler_job)
+ scheduler_job.state = State.RUNNING
+ scheduler_job.hostname = "HOSTNAME"
+ session.add(scheduler_job)
+ self.scheduler_jobs.append(scheduler_job)
+ self.job_runners.append(job_runner)
+ session.commit()
+ scheduler_job.heartbeat(heartbeat_callback=job_runner.heartbeat_callback)
+
+ def _setup_should_report_success_for_ha_schedulers(self, session=None):
+ for _ in range(3):
+ scheduler_job = Job()
+ job_runner = SchedulerJobRunner(job=scheduler_job)
+ scheduler_job.state = State.RUNNING
+ session.add(scheduler_job)
+ self.scheduler_jobs.append(scheduler_job)
+ self.job_runners.append(job_runner)
+ session.commit()
+ scheduler_job.heartbeat(heartbeat_callback=job_runner.heartbeat_callback)
+
+ def _setup_should_ignore_not_running_jobs(self, session=None):
+ for _ in range(3):
+ scheduler_job = Job()
+ job_runner = SchedulerJobRunner(job=scheduler_job)
+ scheduler_job.state = JobState.FAILED
+ session.add(scheduler_job)
+ self.scheduler_jobs.append(scheduler_job)
+ self.job_runners.append(job_runner)
+ session.commit()
+
+ def _setup_should_raise_exception_for_multiple_scheduler_on_one_host(self, session=None):
+ for _ in range(3):
+ scheduler_job = Job()
+ job_runner = SchedulerJobRunner(job=scheduler_job)
+ job_runner.job = scheduler_job
+ scheduler_job.state = State.RUNNING
+ scheduler_job.hostname = "HOSTNAME"
+ session.add(scheduler_job)
+ self.scheduler_jobs.append(scheduler_job)
+ self.job_runners.append(job_runner)
+ session.commit()
+ scheduler_job.heartbeat(heartbeat_callback=job_runner.heartbeat_callback)
+
+ @provide_session
+ def setup(self, testcase: TESTCASE_TYPE, session=None) -> None:
+ """
+ Setup testcase at runtime based on the `testcase` provided by `pytest.mark.parametrize`.
+ """
+ clear_db_jobs()
+ self.scheduler_jobs = []
+ self.job_runners = []
+ setup_method = getattr(self, f"_setup_{testcase}")
+ setup_method(session)
+
+ def teardown_method(self) -> None:
+ if self.job_runners:
+ for job_runner in self.job_runners:
+ if job_runner.processor_agent:
+ job_runner.processor_agent.end()
+ clear_db_jobs()
+
+
+class TestGetJobs(TestJobEndpoint):
+ @pytest.mark.parametrize(
+ "testcase, query_params, expected_status_code, expected_total_entries",
+ [
+ # original testcases refactor from tests/cli/commands/test_jobs_command.py
+ (TESTCASE_ONE_SCHEDULER, {}, 200, 1),
+ (TESTCASE_ONE_SCHEDULER_WITH_HOSTNAME, {"hostname": "HOSTNAME"}, 200, 1),
+ (TESTCASE_HA_SCHEDULERS, {"limit": 100}, 200, 3),
+ (TESTCASE_IGNORE_NOT_RUNNING, {}, 200, 0),
+ (TESTCASE_MULTIPLE_SCHEDULERS_ON_ONE_HOST, {"limit": 100}, 200, 3),
+ ],
+ )
+ def test_get_jobs(
+ self, test_client, testcase, query_params, expected_status_code, expected_total_entries
+ ):
+ # setup testcase at runtime based on the `testcase` parameter
+ self.setup(testcase)
+ response = test_client.get("/public/jobs/", params=query_params)
+ assert response.status_code == expected_status_code
+ if expected_status_code != 200:
+ return
+ response_json = response.json()
+ assert response_json["total_entries"] == expected_total_entries
+
+ for idx, resp_job in enumerate(response_json["jobs"]):
+ expected_job = {
+ "id": self.scheduler_jobs[idx].id,
+ "dag_id": None,
+ "state": "running",
+ "job_type": "SchedulerJob",
+ "start_date": self.scheduler_jobs[idx].start_date.isoformat().replace("+00:00", "Z"),
Review Comment:
Nit: maybe we can use a string formatter to get the `Zulu's` string format, instead of a hard replace like this which does not seem really resilient.
##########
airflow/api_fastapi/core_api/routes/public/job.py:
##########
@@ -0,0 +1,128 @@
+# 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 fastapi import Depends, status
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+from typing_extensions import Annotated
+
+from airflow.api_fastapi.common.db.common import (
+ get_session,
+ paginated_select,
+)
+from airflow.api_fastapi.common.parameters import (
+ QueryLimit,
+ QueryOffset,
+ RangeFilter,
+ SortParam,
+ datetime_range_filter_factory,
+)
+from airflow.api_fastapi.common.router import AirflowRouter
+from airflow.api_fastapi.core_api.datamodels.job import (
+ JobCollectionResponse,
+ JobResponse,
+)
+from airflow.api_fastapi.core_api.openapi.exceptions import create_openapi_http_exception_doc
+from airflow.jobs.job import Job
+from airflow.utils.state import JobState
+
+job_router = AirflowRouter(tags=["Job"], prefix="/jobs")
+
+
+@job_router.get(
+ "/",
+ responses=create_openapi_http_exception_doc(
+ [status.HTTP_400_BAD_REQUEST, status.HTTP_401_UNAUTHORIZED, status.HTTP_403_FORBIDDEN]
+ ),
+)
+def get_jobs(
+ start_date_range: Annotated[
+ RangeFilter,
+ Depends(datetime_range_filter_factory("start_date", Job)),
+ ],
+ end_date_range: Annotated[
+ RangeFilter,
+ Depends(datetime_range_filter_factory("end_date", Job)),
+ ],
+ limit: QueryLimit,
+ offset: QueryOffset,
+ order_by: Annotated[
+ SortParam,
+ Depends(
+ SortParam(
+ [
+ "id",
+ "dag_id",
+ "state",
+ "job_type",
+ "start_date",
+ "end_date",
+ "latest_heartbeat",
+ "executor_class",
+ "hostname",
+ "unixname",
+ ],
+ Job,
+ ).dynamic_depends(default="id")
+ ),
+ ],
+ session: Annotated[Session, Depends(get_session)],
+ state: str | None = None,
+ job_type: str | None = None,
+ hostname: str | None = None,
+ executor_class: str | None = None,
+ is_alive: bool | None = None,
+) -> JobCollectionResponse:
+ """Get all jobs."""
+ base_select = select(Job).where(Job.state == JobState.RUNNING).order_by(Job.latest_heartbeat.desc())
+ # TODO: Refactor using the `FilterParam` class in commit `574b72e41cc5ed175a2bbf4356522589b836bb11`
+ if state:
+ base_select = base_select.where(Job.state == state)
+ if job_type:
+ base_select = base_select.where(Job.job_type == job_type)
+ if hostname:
+ base_select = base_select.where(Job.hostname == hostname)
+ if executor_class:
+ base_select = base_select.where(Job.executor_class == executor_class)
Review Comment:
This needs to be done with standard our standard parameter way and given to `paginated_select`
--
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] AIP-84 Add Lists Jobs with Filters API [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #43859:
URL: https://github.com/apache/airflow/pull/43859#discussion_r1837778262
##########
airflow/api_fastapi/core_api/routes/public/job.py:
##########
@@ -0,0 +1,128 @@
+# 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 fastapi import Depends, status
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+from typing_extensions import Annotated
+
+from airflow.api_fastapi.common.db.common import (
+ get_session,
+ paginated_select,
+)
+from airflow.api_fastapi.common.parameters import (
+ QueryLimit,
+ QueryOffset,
+ RangeFilter,
+ SortParam,
+ datetime_range_filter_factory,
+)
+from airflow.api_fastapi.common.router import AirflowRouter
+from airflow.api_fastapi.core_api.datamodels.job import (
+ JobCollectionResponse,
+ JobResponse,
+)
+from airflow.api_fastapi.core_api.openapi.exceptions import create_openapi_http_exception_doc
+from airflow.jobs.job import Job
+from airflow.utils.state import JobState
+
+job_router = AirflowRouter(tags=["Job"], prefix="/jobs")
+
+
+@job_router.get(
+ "/",
+ responses=create_openapi_http_exception_doc(
+ [status.HTTP_400_BAD_REQUEST, status.HTTP_401_UNAUTHORIZED, status.HTTP_403_FORBIDDEN]
+ ),
+)
+def get_jobs(
+ start_date_range: Annotated[
+ RangeFilter,
+ Depends(datetime_range_filter_factory("start_date", Job)),
+ ],
+ end_date_range: Annotated[
+ RangeFilter,
+ Depends(datetime_range_filter_factory("end_date", Job)),
+ ],
+ limit: QueryLimit,
+ offset: QueryOffset,
+ order_by: Annotated[
+ SortParam,
+ Depends(
+ SortParam(
+ [
+ "id",
+ "dag_id",
+ "state",
+ "job_type",
+ "start_date",
+ "end_date",
+ "latest_heartbeat",
+ "executor_class",
+ "hostname",
+ "unixname",
+ ],
+ Job,
+ ).dynamic_depends(default="id")
+ ),
+ ],
+ session: Annotated[Session, Depends(get_session)],
+ state: str | None = None,
+ job_type: str | None = None,
+ hostname: str | None = None,
+ executor_class: str | None = None,
+ is_alive: bool | None = None,
+) -> JobCollectionResponse:
+ """Get all jobs."""
+ base_select = select(Job).where(Job.state == JobState.RUNNING).order_by(Job.latest_heartbeat.desc())
+ # TODO: Refactor using the `FilterParam` class in commit `574b72e41cc5ed175a2bbf4356522589b836bb11`
+ if state:
+ base_select = base_select.where(Job.state == state)
+ if job_type:
+ base_select = base_select.where(Job.job_type == job_type)
+ if hostname:
+ base_select = base_select.where(Job.hostname == hostname)
+ if executor_class:
+ base_select = base_select.where(Job.executor_class == executor_class)
Review Comment:
This needs to be done with our standard parameters and given to `paginated_select` as we do for other 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] AIP-84 Add Lists Jobs with Filters API [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on PR #43859:
URL: https://github.com/apache/airflow/pull/43859#issuecomment-2493518221
Well I just tried the `conflict` resolution via the github interface. (because it was a really small conflict), quite handy for quick and simple conflicts.
--
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] AIP-84 Add Lists Jobs with Filters API [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on PR #43859:
URL: https://github.com/apache/airflow/pull/43859#issuecomment-2493712020
Hi @pierrejeambrun , I have ran the pre-commit locally and didn't not encounter current CI issue, not sure about what cause the `Generate the FastAPI API spec` error.
--
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] AIP-84 Add Lists Jobs with Filters API [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun merged PR #43859:
URL: https://github.com/apache/airflow/pull/43859
--
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] AIP-84 Add Lists Jobs with Filters API [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #43859:
URL: https://github.com/apache/airflow/pull/43859#discussion_r1837926133
##########
tests/api_fastapi/core_api/routes/public/test_job.py:
##########
@@ -0,0 +1,166 @@
+# 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 typing import Literal
+
+import pytest
+
+from airflow.jobs.job import Job
+from airflow.jobs.scheduler_job_runner import SchedulerJobRunner
+from airflow.utils.session import provide_session
+from airflow.utils.state import JobState, State
+
+from tests_common.test_utils.db import clear_db_jobs
+
+pytestmark = pytest.mark.db_test
+
+TESTCASE_TYPE = Literal[
+ "should_report_success_for_one_working_scheduler",
+ "should_report_success_for_one_working_scheduler_with_hostname",
+ "should_report_success_for_ha_schedulers",
+ "should_ignore_not_running_jobs",
+ "should_raise_exception_for_multiple_scheduler_on_one_host",
+]
+TESTCASE_ONE_SCHEDULER = "should_report_success_for_one_working_scheduler"
+TESTCASE_ONE_SCHEDULER_WITH_HOSTNAME = "should_report_success_for_one_working_scheduler_with_hostname"
+TESTCASE_HA_SCHEDULERS = "should_report_success_for_ha_schedulers"
+TESTCASE_IGNORE_NOT_RUNNING = "should_ignore_not_running_jobs"
+TESTCASE_MULTIPLE_SCHEDULERS_ON_ONE_HOST = "should_raise_exception_for_multiple_scheduler_on_one_host"
+TESTCASE_MULTIPLE_RUNNER = "should_report_success_for_multiple_runners"
+
+
+class TestJobEndpoint:
+ """Common class for /public/jobs related unit tests."""
+
+ scheduler_jobs: list[Job] | None = None
+ job_runners: list[SchedulerJobRunner] | None = None
+
+ def _setup_should_report_success_for_one_working_scheduler(self, session=None):
+ scheduler_job = Job()
+ job_runner = SchedulerJobRunner(job=scheduler_job)
+ scheduler_job.state = State.RUNNING
+ session.add(scheduler_job)
+ session.commit()
+ self.scheduler_jobs.append(scheduler_job)
+ self.job_runners.append(job_runner)
+ scheduler_job.heartbeat(heartbeat_callback=job_runner.heartbeat_callback)
+
+ def _setup_should_report_success_for_one_working_scheduler_with_hostname(self, session=None):
+ scheduler_job = Job()
+ job_runner = SchedulerJobRunner(job=scheduler_job)
+ scheduler_job.state = State.RUNNING
+ scheduler_job.hostname = "HOSTNAME"
+ session.add(scheduler_job)
+ self.scheduler_jobs.append(scheduler_job)
+ self.job_runners.append(job_runner)
+ session.commit()
+ scheduler_job.heartbeat(heartbeat_callback=job_runner.heartbeat_callback)
+
+ def _setup_should_report_success_for_ha_schedulers(self, session=None):
+ for _ in range(3):
+ scheduler_job = Job()
+ job_runner = SchedulerJobRunner(job=scheduler_job)
+ scheduler_job.state = State.RUNNING
+ session.add(scheduler_job)
+ self.scheduler_jobs.append(scheduler_job)
+ self.job_runners.append(job_runner)
+ session.commit()
+ scheduler_job.heartbeat(heartbeat_callback=job_runner.heartbeat_callback)
+
+ def _setup_should_ignore_not_running_jobs(self, session=None):
+ for _ in range(3):
+ scheduler_job = Job()
+ job_runner = SchedulerJobRunner(job=scheduler_job)
+ scheduler_job.state = JobState.FAILED
+ session.add(scheduler_job)
+ self.scheduler_jobs.append(scheduler_job)
+ self.job_runners.append(job_runner)
+ session.commit()
+
+ def _setup_should_raise_exception_for_multiple_scheduler_on_one_host(self, session=None):
+ for _ in range(3):
+ scheduler_job = Job()
+ job_runner = SchedulerJobRunner(job=scheduler_job)
+ job_runner.job = scheduler_job
+ scheduler_job.state = State.RUNNING
+ scheduler_job.hostname = "HOSTNAME"
+ session.add(scheduler_job)
+ self.scheduler_jobs.append(scheduler_job)
+ self.job_runners.append(job_runner)
+ session.commit()
+ scheduler_job.heartbeat(heartbeat_callback=job_runner.heartbeat_callback)
+
+ @provide_session
+ def setup(self, testcase: TESTCASE_TYPE, session=None) -> None:
+ """
+ Setup testcase at runtime based on the `testcase` provided by `pytest.mark.parametrize`.
+ """
+ clear_db_jobs()
+ self.scheduler_jobs = []
+ self.job_runners = []
+ setup_method = getattr(self, f"_setup_{testcase}")
+ setup_method(session)
+
+ def teardown_method(self) -> None:
+ if self.job_runners:
+ for job_runner in self.job_runners:
+ if job_runner.processor_agent:
+ job_runner.processor_agent.end()
+ clear_db_jobs()
+
+
+class TestGetJobs(TestJobEndpoint):
+ @pytest.mark.parametrize(
+ "testcase, query_params, expected_status_code, expected_total_entries",
+ [
+ # original testcases refactor from tests/cli/commands/test_jobs_command.py
+ (TESTCASE_ONE_SCHEDULER, {}, 200, 1),
+ (TESTCASE_ONE_SCHEDULER_WITH_HOSTNAME, {"hostname": "HOSTNAME"}, 200, 1),
+ (TESTCASE_HA_SCHEDULERS, {"limit": 100}, 200, 3),
+ (TESTCASE_IGNORE_NOT_RUNNING, {}, 200, 0),
+ (TESTCASE_MULTIPLE_SCHEDULERS_ON_ONE_HOST, {"limit": 100}, 200, 3),
+ ],
+ )
+ def test_get_jobs(
+ self, test_client, testcase, query_params, expected_status_code, expected_total_entries
+ ):
+ # setup testcase at runtime based on the `testcase` parameter
+ self.setup(testcase)
+ response = test_client.get("/public/jobs/", params=query_params)
+ assert response.status_code == expected_status_code
+ if expected_status_code != 200:
+ return
+ response_json = response.json()
+ assert response_json["total_entries"] == expected_total_entries
+
+ for idx, resp_job in enumerate(response_json["jobs"]):
+ expected_job = {
+ "id": self.scheduler_jobs[idx].id,
+ "dag_id": None,
+ "state": "running",
+ "job_type": "SchedulerJob",
+ "start_date": self.scheduler_jobs[idx].start_date.isoformat().replace("+00:00", "Z"),
Review Comment:
Sure, but is there a string formatting utility in the current testing utils? (I haven't found one in the API test cases.) Or should I add one to the testing utils?
--
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] AIP-84 Add Lists Jobs with Filters API [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #43859:
URL: https://github.com/apache/airflow/pull/43859#discussion_r1838208025
##########
airflow/api_fastapi/core_api/routes/public/job.py:
##########
@@ -0,0 +1,128 @@
+# 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 fastapi import Depends, status
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+from typing_extensions import Annotated
+
+from airflow.api_fastapi.common.db.common import (
+ get_session,
+ paginated_select,
+)
+from airflow.api_fastapi.common.parameters import (
+ QueryLimit,
+ QueryOffset,
+ RangeFilter,
+ SortParam,
+ datetime_range_filter_factory,
+)
+from airflow.api_fastapi.common.router import AirflowRouter
+from airflow.api_fastapi.core_api.datamodels.job import (
+ JobCollectionResponse,
+ JobResponse,
+)
+from airflow.api_fastapi.core_api.openapi.exceptions import create_openapi_http_exception_doc
+from airflow.jobs.job import Job
+from airflow.utils.state import JobState
+
+job_router = AirflowRouter(tags=["Job"], prefix="/jobs")
+
+
+@job_router.get(
+ "/",
+ responses=create_openapi_http_exception_doc(
+ [status.HTTP_400_BAD_REQUEST, status.HTTP_401_UNAUTHORIZED, status.HTTP_403_FORBIDDEN]
+ ),
+)
+def get_jobs(
+ start_date_range: Annotated[
+ RangeFilter,
+ Depends(datetime_range_filter_factory("start_date", Job)),
+ ],
+ end_date_range: Annotated[
+ RangeFilter,
+ Depends(datetime_range_filter_factory("end_date", Job)),
+ ],
+ limit: QueryLimit,
+ offset: QueryOffset,
+ order_by: Annotated[
+ SortParam,
+ Depends(
+ SortParam(
+ [
+ "id",
+ "dag_id",
+ "state",
+ "job_type",
+ "start_date",
+ "end_date",
+ "latest_heartbeat",
+ "executor_class",
+ "hostname",
+ "unixname",
+ ],
+ Job,
+ ).dynamic_depends(default="id")
+ ),
+ ],
+ session: Annotated[Session, Depends(get_session)],
+ state: str | None = None,
+ job_type: str | None = None,
+ hostname: str | None = None,
+ executor_class: str | None = None,
+ is_alive: bool | None = None,
+) -> JobCollectionResponse:
+ """Get all jobs."""
+ base_select = select(Job).where(Job.state == JobState.RUNNING).order_by(Job.latest_heartbeat.desc())
+ # TODO: Refactor using the `FilterParam` class in commit `574b72e41cc5ed175a2bbf4356522589b836bb11`
+ if state:
+ base_select = base_select.where(Job.state == state)
+ if job_type:
+ base_select = base_select.where(Job.job_type == job_type)
+ if hostname:
+ base_select = base_select.where(Job.hostname == hostname)
+ if executor_class:
+ base_select = base_select.where(Job.executor_class == executor_class)
Review Comment:
If we want to merge this before the parameter refactoring is finalized then we need to make this PR consistent and self sufficient.
In case the refactoring PR is delayed, not done, or we change our mind about it, code in main should be clean. This is why I recommend extending the current parameters that we have and use them so we can get this merged.
Then we can change all that with the PR improving the parameter system.
--
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