You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "prabhusneha (via GitHub)" <gi...@apache.org> on 2024/12/03 22:39:17 UTC
[PR] Add ability to list active backfills both across cluster and per-dag [airflow]
prabhusneha opened a new pull request, #44624:
URL: https://github.com/apache/airflow/pull/44624
Related [#44396](https://github.com/apache/airflow/issues/44396)
--
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] Add ability to list active backfills both across cluster and per-dag [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun merged PR #44624:
URL: https://github.com/apache/airflow/pull/44624
--
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] Add ability to list active backfills both across cluster and per-dag [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #44624:
URL: https://github.com/apache/airflow/pull/44624#discussion_r1878296808
##########
airflow/api_fastapi/core_api/routes/ui/backfills.py:
##########
@@ -0,0 +1,66 @@
+# 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 Annotated
+
+from fastapi import Depends, status
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from airflow.api_fastapi.common.db.common import get_async_session, paginated_select_async
+from airflow.api_fastapi.common.parameters import QueryLimit, QueryOffset, SortParam
+from airflow.api_fastapi.common.router import AirflowRouter
+from airflow.api_fastapi.core_api.datamodels.backfills import BackfillCollectionResponse
+from airflow.api_fastapi.core_api.openapi.exceptions import (
+ create_openapi_http_exception_doc,
+)
+from airflow.models.backfill import Backfill
+
+backfills_router = AirflowRouter(tags=["Backfill"], prefix="/backfills")
+
+
+@backfills_router.get(
+ path="",
+ responses=create_openapi_http_exception_doc([status.HTTP_404_NOT_FOUND]),
+)
+async def list_backfills(
+ limit: QueryLimit,
+ offset: QueryOffset,
+ order_by: Annotated[
+ SortParam,
+ Depends(SortParam(["id"], Backfill).dynamic_depends()),
+ ],
+ session: Annotated[AsyncSession, Depends(get_async_session)],
+ dag_id: str | None = None,
+) -> BackfillCollectionResponse:
Review Comment:
We can start like this, and later integrate it to the public API if we need that. Indeed 'active' for a backfill is not really defined at the moment.
--
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] Add ability to list active backfills both across cluster and per-dag [airflow]
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #44624:
URL: https://github.com/apache/airflow/pull/44624#discussion_r1876515016
##########
airflow/api_fastapi/core_api/routes/ui/backfills.py:
##########
@@ -0,0 +1,66 @@
+# 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 Annotated
+
+from fastapi import Depends, status
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from airflow.api_fastapi.common.db.common import get_async_session, paginated_select_async
+from airflow.api_fastapi.common.parameters import QueryLimit, QueryOffset, SortParam
+from airflow.api_fastapi.common.router import AirflowRouter
+from airflow.api_fastapi.core_api.datamodels.backfills import BackfillCollectionResponse
+from airflow.api_fastapi.core_api.openapi.exceptions import (
+ create_openapi_http_exception_doc,
+)
+from airflow.models.backfill import Backfill
+
+backfills_router = AirflowRouter(tags=["Backfill"], prefix="/backfills")
+
+
+@backfills_router.get(
+ path="",
+ responses=create_openapi_http_exception_doc([status.HTTP_404_NOT_FOUND]),
+)
+async def list_backfills(
+ limit: QueryLimit,
+ offset: QueryOffset,
+ order_by: Annotated[
+ SortParam,
+ Depends(SortParam(["id"], Backfill).dynamic_depends()),
+ ],
+ session: Annotated[AsyncSession, Depends(get_async_session)],
+ dag_id: str | None = None,
+) -> BackfillCollectionResponse:
Review Comment:
my feelings on it are not super strong. the thing i hesitate on a bit is like there's no status field for backfills. so "active" is just define by not completed. and i hesitate to add "active" as a filter on the public api.
anyway, having something in private makes it easier to add or make changes there, and avoids us falling into the trap of, increasing the public interface every time the webserver needs something. i like the idea of thinking through the interfaces independently. but that's just me so, i don't stand in the way if the decision is to do something public.
--
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] Add ability to list active backfills both across cluster and per-dag [airflow]
Posted by "prabhusneha (via GitHub)" <gi...@apache.org>.
prabhusneha commented on code in PR #44624:
URL: https://github.com/apache/airflow/pull/44624#discussion_r1876354105
##########
airflow/api_fastapi/core_api/routes/ui/backfills.py:
##########
@@ -0,0 +1,69 @@
+# 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 Annotated
+
+from fastapi import Depends, status
+from sqlalchemy import and_, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from airflow.api_fastapi.common.db.common import get_async_session, paginated_select_async
+from airflow.api_fastapi.common.parameters import QueryLimit, QueryOffset, SortParam
+from airflow.api_fastapi.common.router import AirflowRouter
+from airflow.api_fastapi.core_api.datamodels.backfills import BackfillCollectionResponse
+from airflow.api_fastapi.core_api.openapi.exceptions import (
+ create_openapi_http_exception_doc,
+)
+from airflow.models.backfill import Backfill
+
+backfills_router = AirflowRouter(tags=["Backfill"], prefix="/backfills")
+
+
+@backfills_router.get(
+ path="",
+ responses=create_openapi_http_exception_doc([status.HTTP_404_NOT_FOUND]),
+)
+async def list_backfills(
+ limit: QueryLimit,
+ offset: QueryOffset,
+ order_by: Annotated[
+ SortParam,
+ Depends(SortParam(["id"], Backfill).dynamic_depends()),
+ ],
+ session: Annotated[AsyncSession, Depends(get_async_session)],
+ dag_id: str | None = None,
+) -> BackfillCollectionResponse:
+ print("dag_id")
+ conditions = [Backfill.completed_at.is_(None)] # Active dag
Review Comment:
Added a filter only_active
--
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] Add ability to list active backfills both across cluster and per-dag [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #44624:
URL: https://github.com/apache/airflow/pull/44624#discussion_r1878347993
##########
airflow/api_fastapi/common/parameters.py:
##########
@@ -112,6 +112,34 @@ def depends(self, only_active: bool = True) -> _OnlyActiveFilter:
return self.set_value(only_active)
+class _NoneFilter(BaseParam[bool]):
+ """check if a column is none or not."""
+
+ def __init__(self, attribute: ColumnElement, value: bool | None = None, skip_none: bool = True) -> None:
+ super().__init__(value, skip_none)
+ self.attribute: ColumnElement = attribute
+ self.value: bool | None = value
+
+ def to_orm(self, select: Select) -> Select:
+ if not self.value and self.skip_none:
+ return select
+ return select.where(self.attribute.is_(None))
+
+ def depends(self, *args: Any, **kwargs: Any) -> Self:
+ raise NotImplementedError("Use search_param_factory instead , depends is not implemented.")
+
+
+def none_param_factory(
+ attribute: ColumnElement,
+ none_filter_name: str,
+ skip_none: bool = True,
+) -> Callable[[bool | None], _NoneFilter]:
+ def depends_none(value: bool | None = Query(alias=none_filter_name, default=None)) -> _NoneFilter:
Review Comment:
Maybe a more general approach is to `active=True` `active=False` `active=None`.
- None is ignored and no filtering happen, returning all resources
- True, only take `active` ones, i.e end date is None
- False only take `inactive` ones, i.e end data is not None
This will allow retrieving only `inactive` backfills (which is not possible at the moment), and without having redundancy between `none` and `false` which are basically doing the same for now.
--
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] Add ability to list active backfills both across cluster and per-dag [airflow]
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #44624:
URL: https://github.com/apache/airflow/pull/44624#discussion_r1868462024
##########
airflow/api_fastapi/core_api/routes/ui/backfills.py:
##########
@@ -0,0 +1,69 @@
+# 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 Annotated
+
+from fastapi import Depends, status
+from sqlalchemy import and_, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from airflow.api_fastapi.common.db.common import get_async_session, paginated_select_async
+from airflow.api_fastapi.common.parameters import QueryLimit, QueryOffset, SortParam
+from airflow.api_fastapi.common.router import AirflowRouter
+from airflow.api_fastapi.core_api.datamodels.backfills import BackfillCollectionResponse
+from airflow.api_fastapi.core_api.openapi.exceptions import (
+ create_openapi_http_exception_doc,
+)
+from airflow.models.backfill import Backfill
+
+backfills_router = AirflowRouter(tags=["Backfill"], prefix="/backfills")
+
+
+@backfills_router.get(
+ path="",
+ responses=create_openapi_http_exception_doc([status.HTTP_404_NOT_FOUND]),
+)
+async def list_backfills(
+ limit: QueryLimit,
+ offset: QueryOffset,
+ order_by: Annotated[
+ SortParam,
+ Depends(SortParam(["id"], Backfill).dynamic_depends()),
+ ],
+ session: Annotated[AsyncSession, Depends(get_async_session)],
+ dag_id: str | None = None,
+) -> BackfillCollectionResponse:
+ print("dag_id")
+ conditions = [Backfill.completed_at.is_(None)] # Active dag
Review Comment:
if this endpoint is just for active backfills, then we should name it `list_active_backfills`
if it can accept a param, then the name is fine.
@bbovenzi opinions? requirements?
--
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] Add ability to list active backfills both across cluster and per-dag [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #44624:
URL: https://github.com/apache/airflow/pull/44624#discussion_r1871111707
##########
airflow/api_fastapi/core_api/routes/ui/backfills.py:
##########
@@ -0,0 +1,66 @@
+# 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 Annotated
+
+from fastapi import Depends, status
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from airflow.api_fastapi.common.db.common import get_async_session, paginated_select_async
+from airflow.api_fastapi.common.parameters import QueryLimit, QueryOffset, SortParam
+from airflow.api_fastapi.common.router import AirflowRouter
+from airflow.api_fastapi.core_api.datamodels.backfills import BackfillCollectionResponse
+from airflow.api_fastapi.core_api.openapi.exceptions import (
+ create_openapi_http_exception_doc,
+)
+from airflow.models.backfill import Backfill
+
+backfills_router = AirflowRouter(tags=["Backfill"], prefix="/backfills")
+
+
+@backfills_router.get(
+ path="",
+ responses=create_openapi_http_exception_doc([status.HTTP_404_NOT_FOUND]),
+)
+async def list_backfills(
+ limit: QueryLimit,
+ offset: QueryOffset,
+ order_by: Annotated[
+ SortParam,
+ Depends(SortParam(["id"], Backfill).dynamic_depends()),
+ ],
+ session: Annotated[AsyncSession, Depends(get_async_session)],
+ dag_id: str | None = None,
+) -> BackfillCollectionResponse:
Review Comment:
Do we need a private UI endpoint for that, or should we complete / improve the public one ?
I am asking because this endpoint is really similar to the public one, by adding `dag_id` and `active` filter we should pretty much get all that we want.
I know that the idea is that if it's private we can do whatever we want without worrying about backward comp, but this feels like very generic / public feature.
##########
airflow/api_fastapi/core_api/routes/ui/backfills.py:
##########
@@ -0,0 +1,66 @@
+# 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 Annotated
+
+from fastapi import Depends, status
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from airflow.api_fastapi.common.db.common import get_async_session, paginated_select_async
+from airflow.api_fastapi.common.parameters import QueryLimit, QueryOffset, SortParam
+from airflow.api_fastapi.common.router import AirflowRouter
+from airflow.api_fastapi.core_api.datamodels.backfills import BackfillCollectionResponse
+from airflow.api_fastapi.core_api.openapi.exceptions import (
+ create_openapi_http_exception_doc,
+)
+from airflow.models.backfill import Backfill
+
+backfills_router = AirflowRouter(tags=["Backfill"], prefix="/backfills")
+
+
+@backfills_router.get(
+ path="",
+ responses=create_openapi_http_exception_doc([status.HTTP_404_NOT_FOUND]),
+)
+async def list_backfills(
+ limit: QueryLimit,
+ offset: QueryOffset,
+ order_by: Annotated[
+ SortParam,
+ Depends(SortParam(["id"], Backfill).dynamic_depends()),
+ ],
+ session: Annotated[AsyncSession, Depends(get_async_session)],
+ dag_id: str | None = None,
+) -> BackfillCollectionResponse:
+ conditions = [Backfill.completed_at.is_(None)] # Active dag
+ if dag_id:
+ conditions.append(Backfill.dag_id == dag_id)
+
+ select_stmt, total_entries = await paginated_select_async(
+ statement=select(Backfill).where(*conditions),
Review Comment:
This would be removed.
##########
tests/api_fastapi/core_api/routes/ui/test_backfills.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 unittest import mock
+
+import pendulum
+import pytest
+
+from airflow.models import DagModel
+from airflow.models.backfill import Backfill
+from airflow.utils import timezone
+from airflow.utils.session import provide_session
+
+from tests_common.test_utils.db import (
+ clear_db_backfills,
+ clear_db_dags,
+ clear_db_runs,
+ clear_db_serialized_dags,
+)
+
+pytestmark = pytest.mark.db_test
+
+
+DAG_ID = "test_dag"
+TASK_ID = "op1"
+DAG2_ID = "test_dag2"
+DAG3_ID = "test_dag3"
+
+
+def _clean_db():
+ clear_db_backfills()
+ clear_db_runs()
+ clear_db_dags()
+ clear_db_serialized_dags()
+
+
+@pytest.fixture(autouse=True)
+def clean_db():
+ _clean_db()
+ yield
+ _clean_db()
+
+
+def to_iso(val):
+ return pendulum.instance(val).to_iso8601_string()
+
+
+class TestBackfillEndpoint:
+ @provide_session
+ def _create_dag_models(self, *, count=3, dag_id_prefix="TEST_DAG", is_paused=False, session=None):
+ dags = []
+ for num in range(1, count + 1):
+ dag_model = DagModel(
+ dag_id=f"{dag_id_prefix}_{num}",
+ fileloc=f"/tmp/dag_{num}.py",
+ is_active=True,
+ timetable_summary="0 0 * * *",
+ is_paused=is_paused,
+ )
+ session.add(dag_model)
+ dags.append(dag_model)
+ return dags
+
+
+class TestListBackfills(TestBackfillEndpoint):
+ def test_list_backfill(self, test_client, session):
+ dags = self._create_dag_models()
+ from_date = timezone.utcnow()
+ to_date = timezone.utcnow()
+ backfill0 = Backfill(
+ dag_id=dags[0].dag_id, from_date=from_date, to_date=to_date, completed_at=timezone.utcnow()
+ )
+ backfill1 = Backfill(dag_id=dags[1].dag_id, from_date=from_date, to_date=to_date)
+ backfill2 = Backfill(dag_id=dags[2].dag_id, from_date=from_date, to_date=to_date, is_paused=True)
+ backfills = [backfill0, backfill1, backfill2]
+ for backfill in backfills:
+ session.add(backfill)
Review Comment:
`session.add_all([...])`
##########
tests/api_fastapi/core_api/routes/ui/test_backfills.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 unittest import mock
+
+import pendulum
+import pytest
+
+from airflow.models import DagModel
+from airflow.models.backfill import Backfill
+from airflow.utils import timezone
+from airflow.utils.session import provide_session
+
+from tests_common.test_utils.db import (
+ clear_db_backfills,
+ clear_db_dags,
+ clear_db_runs,
+ clear_db_serialized_dags,
+)
+
+pytestmark = pytest.mark.db_test
+
+
+DAG_ID = "test_dag"
+TASK_ID = "op1"
+DAG2_ID = "test_dag2"
+DAG3_ID = "test_dag3"
+
+
+def _clean_db():
+ clear_db_backfills()
+ clear_db_runs()
+ clear_db_dags()
+ clear_db_serialized_dags()
+
+
+@pytest.fixture(autouse=True)
+def clean_db():
+ _clean_db()
+ yield
+ _clean_db()
+
+
+def to_iso(val):
+ return pendulum.instance(val).to_iso8601_string()
Review Comment:
We have now `from_datetime_to_zulu` and `from_datetime_to_zulu_without_ms` to achieve the same, with is shared test code.
##########
tests/api_fastapi/core_api/routes/ui/test_backfills.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 unittest import mock
+
+import pendulum
+import pytest
+
+from airflow.models import DagModel
+from airflow.models.backfill import Backfill
+from airflow.utils import timezone
+from airflow.utils.session import provide_session
+
+from tests_common.test_utils.db import (
+ clear_db_backfills,
+ clear_db_dags,
+ clear_db_runs,
+ clear_db_serialized_dags,
+)
+
+pytestmark = pytest.mark.db_test
+
+
+DAG_ID = "test_dag"
+TASK_ID = "op1"
+DAG2_ID = "test_dag2"
+DAG3_ID = "test_dag3"
+
+
+def _clean_db():
+ clear_db_backfills()
+ clear_db_runs()
+ clear_db_dags()
+ clear_db_serialized_dags()
+
+
+@pytest.fixture(autouse=True)
+def clean_db():
+ _clean_db()
+ yield
+ _clean_db()
+
+
+def to_iso(val):
+ return pendulum.instance(val).to_iso8601_string()
+
+
+class TestBackfillEndpoint:
+ @provide_session
+ def _create_dag_models(self, *, count=3, dag_id_prefix="TEST_DAG", is_paused=False, session=None):
+ dags = []
+ for num in range(1, count + 1):
+ dag_model = DagModel(
+ dag_id=f"{dag_id_prefix}_{num}",
+ fileloc=f"/tmp/dag_{num}.py",
+ is_active=True,
+ timetable_summary="0 0 * * *",
+ is_paused=is_paused,
+ )
+ session.add(dag_model)
+ dags.append(dag_model)
+ return dags
+
+
+class TestListBackfills(TestBackfillEndpoint):
+ def test_list_backfill(self, test_client, session):
+ dags = self._create_dag_models()
+ from_date = timezone.utcnow()
+ to_date = timezone.utcnow()
+ backfill0 = Backfill(
+ dag_id=dags[0].dag_id, from_date=from_date, to_date=to_date, completed_at=timezone.utcnow()
+ )
+ backfill1 = Backfill(dag_id=dags[1].dag_id, from_date=from_date, to_date=to_date)
+ backfill2 = Backfill(dag_id=dags[2].dag_id, from_date=from_date, to_date=to_date, is_paused=True)
+ backfills = [backfill0, backfill1, backfill2]
+ for backfill in backfills:
+ session.add(backfill)
+ session.commit()
+ response = test_client.get("/ui/backfills?dag_id=")
Review Comment:
We should both test that when `dag_id` is provided, or without `dag_id` provided we get what we want.
i.e parametrize the test on the `params` and use `test_client.get("/ui/backfills", params=params)` (leave url / query param encoding to starlette test client, it's more readable on less prone to encoding error)
##########
airflow/api_fastapi/core_api/routes/ui/backfills.py:
##########
@@ -0,0 +1,66 @@
+# 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 Annotated
+
+from fastapi import Depends, status
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from airflow.api_fastapi.common.db.common import get_async_session, paginated_select_async
+from airflow.api_fastapi.common.parameters import QueryLimit, QueryOffset, SortParam
+from airflow.api_fastapi.common.router import AirflowRouter
+from airflow.api_fastapi.core_api.datamodels.backfills import BackfillCollectionResponse
+from airflow.api_fastapi.core_api.openapi.exceptions import (
+ create_openapi_http_exception_doc,
+)
+from airflow.models.backfill import Backfill
+
+backfills_router = AirflowRouter(tags=["Backfill"], prefix="/backfills")
+
+
+@backfills_router.get(
+ path="",
+ responses=create_openapi_http_exception_doc([status.HTTP_404_NOT_FOUND]),
+)
+async def list_backfills(
+ limit: QueryLimit,
+ offset: QueryOffset,
+ order_by: Annotated[
+ SortParam,
+ Depends(SortParam(["id"], Backfill).dynamic_depends()),
+ ],
+ session: Annotated[AsyncSession, Depends(get_async_session)],
+ dag_id: str | None = None,
+) -> BackfillCollectionResponse:
+ conditions = [Backfill.completed_at.is_(None)] # Active dag
+ if dag_id:
+ conditions.append(Backfill.dag_id == dag_id)
Review Comment:
You should probably reuse the `parameters.py` system that we have, this way you do not have to do any filtering by hands.
##########
airflow/api_fastapi/core_api/routes/ui/backfills.py:
##########
@@ -0,0 +1,69 @@
+# 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 Annotated
+
+from fastapi import Depends, status
+from sqlalchemy import and_, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from airflow.api_fastapi.common.db.common import get_async_session, paginated_select_async
+from airflow.api_fastapi.common.parameters import QueryLimit, QueryOffset, SortParam
+from airflow.api_fastapi.common.router import AirflowRouter
+from airflow.api_fastapi.core_api.datamodels.backfills import BackfillCollectionResponse
+from airflow.api_fastapi.core_api.openapi.exceptions import (
+ create_openapi_http_exception_doc,
+)
+from airflow.models.backfill import Backfill
+
+backfills_router = AirflowRouter(tags=["Backfill"], prefix="/backfills")
+
+
+@backfills_router.get(
+ path="",
+ responses=create_openapi_http_exception_doc([status.HTTP_404_NOT_FOUND]),
+)
+async def list_backfills(
+ limit: QueryLimit,
+ offset: QueryOffset,
+ order_by: Annotated[
+ SortParam,
+ Depends(SortParam(["id"], Backfill).dynamic_depends()),
+ ],
+ session: Annotated[AsyncSession, Depends(get_async_session)],
+ dag_id: str | None = None,
+) -> BackfillCollectionResponse:
+ print("dag_id")
+ conditions = [Backfill.completed_at.is_(None)] # Active dag
Review Comment:
Yes I think we can simply add a filter to retrieve all backfills or only `active` ones.
--
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] Add ability to list active backfills both across cluster and per-dag [airflow]
Posted by "prabhusneha (via GitHub)" <gi...@apache.org>.
prabhusneha commented on code in PR #44624:
URL: https://github.com/apache/airflow/pull/44624#discussion_r1868776195
##########
airflow/api_fastapi/core_api/datamodels/ui/backfills.py:
##########
@@ -0,0 +1,46 @@
+# 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
+
+from airflow.api_fastapi.core_api.base import BaseModel
+from airflow.models.backfill import ReprocessBehavior
+
+
+class BackfillResponse(BaseModel):
Review Comment:
Right. Modified this.
--
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] Add ability to list active backfills both across cluster and per-dag [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on PR #44624:
URL: https://github.com/apache/airflow/pull/44624#issuecomment-2519920462
CI need fixing I think.
--
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] Add ability to list active backfills both across cluster and per-dag [airflow]
Posted by "prabhusneha (via GitHub)" <gi...@apache.org>.
prabhusneha commented on code in PR #44624:
URL: https://github.com/apache/airflow/pull/44624#discussion_r1876355279
##########
tests/api_fastapi/core_api/routes/ui/test_backfills.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 unittest import mock
+
+import pendulum
+import pytest
+
+from airflow.models import DagModel
+from airflow.models.backfill import Backfill
+from airflow.utils import timezone
+from airflow.utils.session import provide_session
+
+from tests_common.test_utils.db import (
+ clear_db_backfills,
+ clear_db_dags,
+ clear_db_runs,
+ clear_db_serialized_dags,
+)
+
+pytestmark = pytest.mark.db_test
+
+
+DAG_ID = "test_dag"
+TASK_ID = "op1"
+DAG2_ID = "test_dag2"
+DAG3_ID = "test_dag3"
+
+
+def _clean_db():
+ clear_db_backfills()
+ clear_db_runs()
+ clear_db_dags()
+ clear_db_serialized_dags()
+
+
+@pytest.fixture(autouse=True)
+def clean_db():
+ _clean_db()
+ yield
+ _clean_db()
+
+
+def to_iso(val):
+ return pendulum.instance(val).to_iso8601_string()
+
+
+class TestBackfillEndpoint:
+ @provide_session
+ def _create_dag_models(self, *, count=3, dag_id_prefix="TEST_DAG", is_paused=False, session=None):
+ dags = []
+ for num in range(1, count + 1):
+ dag_model = DagModel(
+ dag_id=f"{dag_id_prefix}_{num}",
+ fileloc=f"/tmp/dag_{num}.py",
+ is_active=True,
+ timetable_summary="0 0 * * *",
+ is_paused=is_paused,
+ )
+ session.add(dag_model)
+ dags.append(dag_model)
+ return dags
+
+
+class TestListBackfills(TestBackfillEndpoint):
+ def test_list_backfill(self, test_client, session):
+ dags = self._create_dag_models()
+ from_date = timezone.utcnow()
+ to_date = timezone.utcnow()
+ backfill0 = Backfill(
+ dag_id=dags[0].dag_id, from_date=from_date, to_date=to_date, completed_at=timezone.utcnow()
+ )
+ backfill1 = Backfill(dag_id=dags[1].dag_id, from_date=from_date, to_date=to_date)
+ backfill2 = Backfill(dag_id=dags[2].dag_id, from_date=from_date, to_date=to_date, is_paused=True)
+ backfills = [backfill0, backfill1, backfill2]
+ for backfill in backfills:
+ session.add(backfill)
+ session.commit()
+ response = test_client.get("/ui/backfills?dag_id=")
Review Comment:
parametrized the test
--
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] Add ability to list active backfills both across cluster and per-dag [airflow]
Posted by "prabhusneha (via GitHub)" <gi...@apache.org>.
prabhusneha commented on code in PR #44624:
URL: https://github.com/apache/airflow/pull/44624#discussion_r1868776415
##########
airflow/api_fastapi/core_api/routes/ui/backfills.py:
##########
@@ -0,0 +1,69 @@
+# 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 Annotated
+
+from fastapi import Depends, status
+from sqlalchemy import and_, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from airflow.api_fastapi.common.db.common import get_async_session, paginated_select_async
+from airflow.api_fastapi.common.parameters import QueryLimit, QueryOffset, SortParam
+from airflow.api_fastapi.common.router import AirflowRouter
+from airflow.api_fastapi.core_api.datamodels.backfills import BackfillCollectionResponse
+from airflow.api_fastapi.core_api.openapi.exceptions import (
+ create_openapi_http_exception_doc,
+)
+from airflow.models.backfill import Backfill
+
+backfills_router = AirflowRouter(tags=["Backfill"], prefix="/backfills")
+
+
+@backfills_router.get(
+ path="",
+ responses=create_openapi_http_exception_doc([status.HTTP_404_NOT_FOUND]),
+)
+async def list_backfills(
+ limit: QueryLimit,
+ offset: QueryOffset,
+ order_by: Annotated[
+ SortParam,
+ Depends(SortParam(["id"], Backfill).dynamic_depends()),
+ ],
+ session: Annotated[AsyncSession, Depends(get_async_session)],
+ dag_id: str | None = None,
+) -> BackfillCollectionResponse:
+ print("dag_id")
+ conditions = [Backfill.completed_at.is_(None)] # Active dag
+ if dag_id:
+ conditions.append(Backfill.dag_id == dag_id)
+ print(dag_id, str(conditions))
+
+ select_stmt, total_entries = await paginated_select_async(
+ statement=select(Backfill).where(and_(*conditions)),
Review Comment:
Modified.
--
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] Add ability to list active backfills both across cluster and per-dag [airflow]
Posted by "prabhusneha (via GitHub)" <gi...@apache.org>.
prabhusneha commented on code in PR #44624:
URL: https://github.com/apache/airflow/pull/44624#discussion_r1876354729
##########
airflow/api_fastapi/core_api/routes/ui/backfills.py:
##########
@@ -0,0 +1,66 @@
+# 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 Annotated
+
+from fastapi import Depends, status
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from airflow.api_fastapi.common.db.common import get_async_session, paginated_select_async
+from airflow.api_fastapi.common.parameters import QueryLimit, QueryOffset, SortParam
+from airflow.api_fastapi.common.router import AirflowRouter
+from airflow.api_fastapi.core_api.datamodels.backfills import BackfillCollectionResponse
+from airflow.api_fastapi.core_api.openapi.exceptions import (
+ create_openapi_http_exception_doc,
+)
+from airflow.models.backfill import Backfill
+
+backfills_router = AirflowRouter(tags=["Backfill"], prefix="/backfills")
+
+
+@backfills_router.get(
+ path="",
+ responses=create_openapi_http_exception_doc([status.HTTP_404_NOT_FOUND]),
+)
+async def list_backfills(
+ limit: QueryLimit,
+ offset: QueryOffset,
+ order_by: Annotated[
+ SortParam,
+ Depends(SortParam(["id"], Backfill).dynamic_depends()),
+ ],
+ session: Annotated[AsyncSession, Depends(get_async_session)],
+ dag_id: str | None = None,
+) -> BackfillCollectionResponse:
+ conditions = [Backfill.completed_at.is_(None)] # Active dag
+ if dag_id:
+ conditions.append(Backfill.dag_id == dag_id)
Review Comment:
Modified the code
--
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] Add ability to list active backfills both across cluster and per-dag [airflow]
Posted by "prabhusneha (via GitHub)" <gi...@apache.org>.
prabhusneha commented on code in PR #44624:
URL: https://github.com/apache/airflow/pull/44624#discussion_r1876355671
##########
tests/api_fastapi/core_api/routes/ui/test_backfills.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 unittest import mock
+
+import pendulum
+import pytest
+
+from airflow.models import DagModel
+from airflow.models.backfill import Backfill
+from airflow.utils import timezone
+from airflow.utils.session import provide_session
+
+from tests_common.test_utils.db import (
+ clear_db_backfills,
+ clear_db_dags,
+ clear_db_runs,
+ clear_db_serialized_dags,
+)
+
+pytestmark = pytest.mark.db_test
+
+
+DAG_ID = "test_dag"
+TASK_ID = "op1"
+DAG2_ID = "test_dag2"
+DAG3_ID = "test_dag3"
+
+
+def _clean_db():
+ clear_db_backfills()
+ clear_db_runs()
+ clear_db_dags()
+ clear_db_serialized_dags()
+
+
+@pytest.fixture(autouse=True)
+def clean_db():
+ _clean_db()
+ yield
+ _clean_db()
+
+
+def to_iso(val):
+ return pendulum.instance(val).to_iso8601_string()
Review Comment:
Changed to this 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] Add ability to list active backfills both across cluster and per-dag [airflow]
Posted by "prabhusneha (via GitHub)" <gi...@apache.org>.
prabhusneha commented on code in PR #44624:
URL: https://github.com/apache/airflow/pull/44624#discussion_r1879697660
##########
airflow/api_fastapi/common/parameters.py:
##########
@@ -112,6 +112,34 @@ def depends(self, only_active: bool = True) -> _OnlyActiveFilter:
return self.set_value(only_active)
+class _NoneFilter(BaseParam[bool]):
+ """check if a column is none or not."""
+
+ def __init__(self, attribute: ColumnElement, value: bool | None = None, skip_none: bool = True) -> None:
+ super().__init__(value, skip_none)
+ self.attribute: ColumnElement = attribute
+ self.value: bool | None = value
+
+ def to_orm(self, select: Select) -> Select:
+ if not self.value and self.skip_none:
+ return select
+ return select.where(self.attribute.is_(None))
+
+ def depends(self, *args: Any, **kwargs: Any) -> Self:
+ raise NotImplementedError("Use search_param_factory instead , depends is not implemented.")
+
+
+def none_param_factory(
+ attribute: ColumnElement,
+ none_filter_name: str,
+ skip_none: bool = True,
+) -> Callable[[bool | None], _NoneFilter]:
+ def depends_none(value: bool | None = Query(alias=none_filter_name, default=None)) -> _NoneFilter:
Review Comment:
Added the 3 cases.
--
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] Add ability to list active backfills both across cluster and per-dag [airflow]
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #44624:
URL: https://github.com/apache/airflow/pull/44624#discussion_r1868460504
##########
airflow/api_fastapi/core_api/routes/ui/backfills.py:
##########
@@ -0,0 +1,69 @@
+# 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 Annotated
+
+from fastapi import Depends, status
+from sqlalchemy import and_, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from airflow.api_fastapi.common.db.common import get_async_session, paginated_select_async
+from airflow.api_fastapi.common.parameters import QueryLimit, QueryOffset, SortParam
+from airflow.api_fastapi.common.router import AirflowRouter
+from airflow.api_fastapi.core_api.datamodels.backfills import BackfillCollectionResponse
+from airflow.api_fastapi.core_api.openapi.exceptions import (
+ create_openapi_http_exception_doc,
+)
+from airflow.models.backfill import Backfill
+
+backfills_router = AirflowRouter(tags=["Backfill"], prefix="/backfills")
+
+
+@backfills_router.get(
+ path="",
+ responses=create_openapi_http_exception_doc([status.HTTP_404_NOT_FOUND]),
+)
+async def list_backfills(
+ limit: QueryLimit,
+ offset: QueryOffset,
+ order_by: Annotated[
+ SortParam,
+ Depends(SortParam(["id"], Backfill).dynamic_depends()),
+ ],
+ session: Annotated[AsyncSession, Depends(get_async_session)],
+ dag_id: str | None = None,
+) -> BackfillCollectionResponse:
+ print("dag_id")
+ conditions = [Backfill.completed_at.is_(None)] # Active dag
+ if dag_id:
+ conditions.append(Backfill.dag_id == dag_id)
+ print(dag_id, str(conditions))
+
+ select_stmt, total_entries = await paginated_select_async(
+ statement=select(Backfill).where(and_(*conditions)),
Review Comment:
i think the and might be redundant; i think where(*conditions) is implicitly and
##########
airflow/api_fastapi/core_api/routes/ui/backfills.py:
##########
@@ -0,0 +1,69 @@
+# 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 Annotated
+
+from fastapi import Depends, status
+from sqlalchemy import and_, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from airflow.api_fastapi.common.db.common import get_async_session, paginated_select_async
+from airflow.api_fastapi.common.parameters import QueryLimit, QueryOffset, SortParam
+from airflow.api_fastapi.common.router import AirflowRouter
+from airflow.api_fastapi.core_api.datamodels.backfills import BackfillCollectionResponse
+from airflow.api_fastapi.core_api.openapi.exceptions import (
+ create_openapi_http_exception_doc,
+)
+from airflow.models.backfill import Backfill
+
+backfills_router = AirflowRouter(tags=["Backfill"], prefix="/backfills")
+
+
+@backfills_router.get(
+ path="",
+ responses=create_openapi_http_exception_doc([status.HTTP_404_NOT_FOUND]),
+)
+async def list_backfills(
+ limit: QueryLimit,
+ offset: QueryOffset,
+ order_by: Annotated[
+ SortParam,
+ Depends(SortParam(["id"], Backfill).dynamic_depends()),
+ ],
+ session: Annotated[AsyncSession, Depends(get_async_session)],
+ dag_id: str | None = None,
+) -> BackfillCollectionResponse:
+ print("dag_id")
+ conditions = [Backfill.completed_at.is_(None)] # Active dag
+ if dag_id:
+ conditions.append(Backfill.dag_id == dag_id)
+ print(dag_id, str(conditions))
+
+ select_stmt, total_entries = await paginated_select_async(
+ statement=select(Backfill).where(and_(*conditions)),
+ order_by=order_by,
+ offset=offset,
+ limit=limit,
+ session=session,
+ )
+ print(select_stmt)
Review Comment:
no print
--
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] Add ability to list active backfills both across cluster and per-dag [airflow]
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #44624:
URL: https://github.com/apache/airflow/pull/44624#discussion_r1868460162
##########
airflow/api_fastapi/core_api/routes/ui/backfills.py:
##########
@@ -0,0 +1,69 @@
+# 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 Annotated
+
+from fastapi import Depends, status
+from sqlalchemy import and_, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from airflow.api_fastapi.common.db.common import get_async_session, paginated_select_async
+from airflow.api_fastapi.common.parameters import QueryLimit, QueryOffset, SortParam
+from airflow.api_fastapi.common.router import AirflowRouter
+from airflow.api_fastapi.core_api.datamodels.backfills import BackfillCollectionResponse
+from airflow.api_fastapi.core_api.openapi.exceptions import (
+ create_openapi_http_exception_doc,
+)
+from airflow.models.backfill import Backfill
+
+backfills_router = AirflowRouter(tags=["Backfill"], prefix="/backfills")
+
+
+@backfills_router.get(
+ path="",
+ responses=create_openapi_http_exception_doc([status.HTTP_404_NOT_FOUND]),
+)
+async def list_backfills(
+ limit: QueryLimit,
+ offset: QueryOffset,
+ order_by: Annotated[
+ SortParam,
+ Depends(SortParam(["id"], Backfill).dynamic_depends()),
+ ],
+ session: Annotated[AsyncSession, Depends(get_async_session)],
+ dag_id: str | None = None,
+) -> BackfillCollectionResponse:
+ print("dag_id")
+ conditions = [Backfill.completed_at.is_(None)] # Active dag
+ if dag_id:
+ conditions.append(Backfill.dag_id == dag_id)
+ print(dag_id, str(conditions))
Review Comment:
don't want to be printing here
--
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] Add ability to list active backfills both across cluster and per-dag [airflow]
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #44624:
URL: https://github.com/apache/airflow/pull/44624#discussion_r1868459830
##########
airflow/api_fastapi/core_api/datamodels/ui/backfills.py:
##########
@@ -0,0 +1,46 @@
+# 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
+
+from airflow.api_fastapi.core_api.base import BaseModel
+from airflow.models.backfill import ReprocessBehavior
+
+
+class BackfillResponse(BaseModel):
Review Comment:
@prabhusneha are these models any different from those in `airflow/api_fastapi/core_api/datamodels/backfills.py`? if not, seems we could just reuse them rather than duplicating?
--
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] Add ability to list active backfills both across cluster and per-dag [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #44624:
URL: https://github.com/apache/airflow/pull/44624#discussion_r1878316280
##########
tests/api_fastapi/core_api/routes/ui/test_backfills.py:
##########
@@ -0,0 +1,153 @@
+# 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 unittest import mock
+
+import pytest
+
+from airflow.models import DagModel
+from airflow.models.backfill import Backfill
+from airflow.utils import timezone
+from airflow.utils.session import provide_session
+
+from tests_common.test_utils.db import (
+ clear_db_backfills,
+ clear_db_dags,
+ clear_db_runs,
+ clear_db_serialized_dags,
+)
+from tests_common.test_utils.format_datetime import from_datetime_to_zulu
+
+pytestmark = pytest.mark.db_test
+
+DAG_ID = "test_dag"
+TASK_ID = "op1"
+DAG2_ID = "test_dag2"
+DAG3_ID = "test_dag3"
+
+
+def _clean_db():
+ clear_db_backfills()
+ clear_db_runs()
+ clear_db_dags()
+ clear_db_serialized_dags()
+
+
+@pytest.fixture(autouse=True)
+def clean_db():
+ _clean_db()
+ yield
+ _clean_db()
+
+
+class TestBackfillEndpoint:
+ @provide_session
+ def _create_dag_models(self, *, count=3, dag_id_prefix="TEST_DAG", is_paused=False, session=None):
+ dags = []
+ for num in range(1, count + 1):
+ dag_model = DagModel(
+ dag_id=f"{dag_id_prefix}_{num}",
+ fileloc=f"/tmp/dag_{num}.py",
+ is_active=True,
+ timetable_summary="0 0 * * *",
+ is_paused=is_paused,
+ )
+ session.add(dag_model)
+ dags.append(dag_model)
+ return dags
+
+
+class TestListBackfills(TestBackfillEndpoint):
+ @pytest.mark.parametrize(
+ "test_params, response_params, total_entries",
+ [
+ ({}, ["backfill1", "backfill2", "backfill3"], 3),
+ ({"only_active": True}, ["backfill2", "backfill3"], 2),
+ ({"only_active": False}, ["backfill1", "backfill2", "backfill3"], 3),
+ ({"dag_id": "", "only_active": True}, ["backfill2", "backfill3"], 2),
+ ({"dag_id": "", "only_active": False}, ["backfill1", "backfill2", "backfill3"], 3),
+ ({"dag_id": ""}, ["backfill1", "backfill2", "backfill3"], 3),
+ ({"dag_id": "TEST_DAG_1", "only_active": True}, [], 0),
+ ({"dag_id": "TEST_DAG_1", "only_active": False}, ["backfill1"], 1),
+ ({"dag_id": "TEST_DAG_1"}, ["backfill1"], 1),
+ ],
+ )
+ @provide_session
Review Comment:
```suggestion
```
##########
airflow/api_fastapi/common/parameters.py:
##########
@@ -112,6 +112,34 @@ def depends(self, only_active: bool = True) -> _OnlyActiveFilter:
return self.set_value(only_active)
+class _NoneFilter(BaseParam[bool]):
+ """check if a column is none or not."""
+
+ def __init__(self, attribute: ColumnElement, value: bool | None = None, skip_none: bool = True) -> None:
+ super().__init__(value, skip_none)
+ self.attribute: ColumnElement = attribute
+ self.value: bool | None = value
+
+ def to_orm(self, select: Select) -> Select:
+ if not self.value and self.skip_none:
+ return select
+ return select.where(self.attribute.is_(None))
+
+ def depends(self, *args: Any, **kwargs: Any) -> Self:
+ raise NotImplementedError("Use search_param_factory instead , depends is not implemented.")
+
+
+def none_param_factory(
+ attribute: ColumnElement,
+ none_filter_name: str,
+ skip_none: bool = True,
+) -> Callable[[bool | None], _NoneFilter]:
+ def depends_none(value: bool | None = Query(alias=none_filter_name, default=None)) -> _NoneFilter:
Review Comment:
```suggestion
def depends_none(value: bool = Query(alias=none_filter_name, default=False)) -> _NoneFilter:
```
`only_active=False` return `all`, same as `only_active=None`. We don't need both `None` and `False` I think
##########
tests/api_fastapi/core_api/routes/ui/test_backfills.py:
##########
@@ -0,0 +1,153 @@
+# 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 unittest import mock
+
+import pytest
+
+from airflow.models import DagModel
+from airflow.models.backfill import Backfill
+from airflow.utils import timezone
+from airflow.utils.session import provide_session
+
+from tests_common.test_utils.db import (
+ clear_db_backfills,
+ clear_db_dags,
+ clear_db_runs,
+ clear_db_serialized_dags,
+)
+from tests_common.test_utils.format_datetime import from_datetime_to_zulu
+
+pytestmark = pytest.mark.db_test
+
+DAG_ID = "test_dag"
+TASK_ID = "op1"
+DAG2_ID = "test_dag2"
+DAG3_ID = "test_dag3"
+
+
+def _clean_db():
+ clear_db_backfills()
+ clear_db_runs()
+ clear_db_dags()
+ clear_db_serialized_dags()
+
+
+@pytest.fixture(autouse=True)
+def clean_db():
+ _clean_db()
+ yield
+ _clean_db()
+
+
+class TestBackfillEndpoint:
+ @provide_session
+ def _create_dag_models(self, *, count=3, dag_id_prefix="TEST_DAG", is_paused=False, session=None):
+ dags = []
+ for num in range(1, count + 1):
+ dag_model = DagModel(
+ dag_id=f"{dag_id_prefix}_{num}",
+ fileloc=f"/tmp/dag_{num}.py",
+ is_active=True,
+ timetable_summary="0 0 * * *",
+ is_paused=is_paused,
+ )
+ session.add(dag_model)
+ dags.append(dag_model)
+ return dags
+
+
+class TestListBackfills(TestBackfillEndpoint):
+ @pytest.mark.parametrize(
+ "test_params, response_params, total_entries",
+ [
+ ({}, ["backfill1", "backfill2", "backfill3"], 3),
+ ({"only_active": True}, ["backfill2", "backfill3"], 2),
+ ({"only_active": False}, ["backfill1", "backfill2", "backfill3"], 3),
+ ({"dag_id": "", "only_active": True}, ["backfill2", "backfill3"], 2),
+ ({"dag_id": "", "only_active": False}, ["backfill1", "backfill2", "backfill3"], 3),
+ ({"dag_id": ""}, ["backfill1", "backfill2", "backfill3"], 3),
+ ({"dag_id": "TEST_DAG_1", "only_active": True}, [], 0),
+ ({"dag_id": "TEST_DAG_1", "only_active": False}, ["backfill1"], 1),
+ ({"dag_id": "TEST_DAG_1"}, ["backfill1"], 1),
+ ],
+ )
+ @provide_session
Review Comment:
session is a fixture now
##########
airflow/api_fastapi/common/parameters.py:
##########
@@ -112,6 +112,34 @@ def depends(self, only_active: bool = True) -> _OnlyActiveFilter:
return self.set_value(only_active)
+class _NoneFilter(BaseParam[bool]):
+ """check if a column is none or not."""
+
+ def __init__(self, attribute: ColumnElement, value: bool | None = None, skip_none: bool = True) -> None:
+ super().__init__(value, skip_none)
+ self.attribute: ColumnElement = attribute
+ self.value: bool | None = value
+
+ def to_orm(self, select: Select) -> Select:
+ if not self.value and self.skip_none:
+ return select
+ return select.where(self.attribute.is_(None))
+
+ def depends(self, *args: Any, **kwargs: Any) -> Self:
+ raise NotImplementedError("Use search_param_factory instead , depends is not implemented.")
+
+
+def none_param_factory(
+ attribute: ColumnElement,
+ none_filter_name: str,
+ skip_none: bool = True,
+) -> Callable[[bool | None], _NoneFilter]:
Review Comment:
Something like (in your case):
```python
if self.filter_option == FilterOptionEnum.IS_NONE:
if self.value == False or self.value is None:
return select
if self.value == True:
return select.where(self.attribute.is_(None))
```
Also this particular enum option is only available with `bool | None` / `bool` type.
##########
airflow/api_fastapi/common/parameters.py:
##########
@@ -112,6 +112,34 @@ def depends(self, only_active: bool = True) -> _OnlyActiveFilter:
return self.set_value(only_active)
+class _NoneFilter(BaseParam[bool]):
+ """check if a column is none or not."""
+
+ def __init__(self, attribute: ColumnElement, value: bool | None = None, skip_none: bool = True) -> None:
+ super().__init__(value, skip_none)
+ self.attribute: ColumnElement = attribute
+ self.value: bool | None = value
+
+ def to_orm(self, select: Select) -> Select:
+ if not self.value and self.skip_none:
+ return select
+ return select.where(self.attribute.is_(None))
+
+ def depends(self, *args: Any, **kwargs: Any) -> Self:
+ raise NotImplementedError("Use search_param_factory instead , depends is not implemented.")
+
+
+def none_param_factory(
+ attribute: ColumnElement,
+ none_filter_name: str,
+ skip_none: bool = True,
+) -> Callable[[bool | None], _NoneFilter]:
+ def depends_none(value: bool | None = Query(alias=none_filter_name, default=None)) -> _NoneFilter:
Review Comment:
Maybe a more general approach is to `active=True` `active=False` `active=None`.
- None is ignored and no filtering happen, returning all resources
- True, only take `active` ones, i.e end date is None
- False only take `inactive` ones, i.e end data is not None
This will allow retrieving only `inactive` backfills (which is not possible at the moment)
##########
airflow/api_fastapi/common/parameters.py:
##########
@@ -112,6 +112,34 @@ def depends(self, only_active: bool = True) -> _OnlyActiveFilter:
return self.set_value(only_active)
+class _NoneFilter(BaseParam[bool]):
+ """check if a column is none or not."""
+
+ def __init__(self, attribute: ColumnElement, value: bool | None = None, skip_none: bool = True) -> None:
+ super().__init__(value, skip_none)
+ self.attribute: ColumnElement = attribute
+ self.value: bool | None = value
+
+ def to_orm(self, select: Select) -> Select:
+ if not self.value and self.skip_none:
+ return select
+ return select.where(self.attribute.is_(None))
+
+ def depends(self, *args: Any, **kwargs: Any) -> Self:
+ raise NotImplementedError("Use search_param_factory instead , depends is not implemented.")
+
+
+def none_param_factory(
+ attribute: ColumnElement,
+ none_filter_name: str,
+ skip_none: bool = True,
+) -> Callable[[bool | None], _NoneFilter]:
Review Comment:
I think all that could be implemented with a `is_none` operator in the `FilterOptionEnum` ? And then using the generic `filter_param_factory` ? Might limit code addition.
--
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] Add ability to list active backfills both across cluster and per-dag [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #44624:
URL: https://github.com/apache/airflow/pull/44624#discussion_r1878347993
##########
airflow/api_fastapi/common/parameters.py:
##########
@@ -112,6 +112,34 @@ def depends(self, only_active: bool = True) -> _OnlyActiveFilter:
return self.set_value(only_active)
+class _NoneFilter(BaseParam[bool]):
+ """check if a column is none or not."""
+
+ def __init__(self, attribute: ColumnElement, value: bool | None = None, skip_none: bool = True) -> None:
+ super().__init__(value, skip_none)
+ self.attribute: ColumnElement = attribute
+ self.value: bool | None = value
+
+ def to_orm(self, select: Select) -> Select:
+ if not self.value and self.skip_none:
+ return select
+ return select.where(self.attribute.is_(None))
+
+ def depends(self, *args: Any, **kwargs: Any) -> Self:
+ raise NotImplementedError("Use search_param_factory instead , depends is not implemented.")
+
+
+def none_param_factory(
+ attribute: ColumnElement,
+ none_filter_name: str,
+ skip_none: bool = True,
+) -> Callable[[bool | None], _NoneFilter]:
+ def depends_none(value: bool | None = Query(alias=none_filter_name, default=None)) -> _NoneFilter:
Review Comment:
Maybe a more general approach is to `active=True` `active=False` `active=None`.
- None is ignored and no filtering happen, returning all resources
- True, only take `active` ones, i.e end date is None
- False only take `inactive` ones, i.e end data is not None
This will allow retrieving only `inactive` backfills (which is not possible at the moment and might be useful for other None filters), and without having redundancy between `none` and `false` which are basically doing the same for now.
##########
airflow/api_fastapi/common/parameters.py:
##########
@@ -112,6 +112,34 @@ def depends(self, only_active: bool = True) -> _OnlyActiveFilter:
return self.set_value(only_active)
+class _NoneFilter(BaseParam[bool]):
+ """check if a column is none or not."""
+
+ def __init__(self, attribute: ColumnElement, value: bool | None = None, skip_none: bool = True) -> None:
+ super().__init__(value, skip_none)
+ self.attribute: ColumnElement = attribute
+ self.value: bool | None = value
+
+ def to_orm(self, select: Select) -> Select:
+ if not self.value and self.skip_none:
+ return select
+ return select.where(self.attribute.is_(None))
+
+ def depends(self, *args: Any, **kwargs: Any) -> Self:
+ raise NotImplementedError("Use search_param_factory instead , depends is not implemented.")
+
+
+def none_param_factory(
+ attribute: ColumnElement,
+ none_filter_name: str,
+ skip_none: bool = True,
+) -> Callable[[bool | None], _NoneFilter]:
+ def depends_none(value: bool | None = Query(alias=none_filter_name, default=None)) -> _NoneFilter:
Review Comment:
Maybe a more general approach is to `active=True` `active=False` `active=None`.
- None is ignored and no filtering happen, returning all resources
- True, only take `active` ones, i.e end date is None
- False only take `inactive` ones, i.e end data is not None
This will allow retrieving only `inactive` backfills (which is not possible at the moment and might be useful for others _NoneFilter filters), and without having redundancy between `none` and `false` which are basically doing the same for now.
--
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] Add ability to list active backfills both across cluster and per-dag [airflow]
Posted by "prabhusneha (via GitHub)" <gi...@apache.org>.
prabhusneha commented on code in PR #44624:
URL: https://github.com/apache/airflow/pull/44624#discussion_r1879697035
##########
airflow/api_fastapi/common/parameters.py:
##########
@@ -112,6 +112,34 @@ def depends(self, only_active: bool = True) -> _OnlyActiveFilter:
return self.set_value(only_active)
+class _NoneFilter(BaseParam[bool]):
+ """check if a column is none or not."""
+
+ def __init__(self, attribute: ColumnElement, value: bool | None = None, skip_none: bool = True) -> None:
+ super().__init__(value, skip_none)
+ self.attribute: ColumnElement = attribute
+ self.value: bool | None = value
+
+ def to_orm(self, select: Select) -> Select:
+ if not self.value and self.skip_none:
+ return select
+ return select.where(self.attribute.is_(None))
+
+ def depends(self, *args: Any, **kwargs: Any) -> Self:
+ raise NotImplementedError("Use search_param_factory instead , depends is not implemented.")
+
+
+def none_param_factory(
+ attribute: ColumnElement,
+ none_filter_name: str,
+ skip_none: bool = True,
+) -> Callable[[bool | None], _NoneFilter]:
Review Comment:
Modified the code
--
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