You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "vatsrahul1001 (via GitHub)" <gi...@apache.org> on 2025/01/17 06:32:59 UTC
[PR] Handle backfill for DAGs with when depends_on_past is True [airflow]
vatsrahul1001 opened a new pull request, #45731:
URL: https://github.com/apache/airflow/pull/45731
<!-- Please keep an empty line above the dashes. -->
---
**^ Add meaningful description above**
Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)** for more information.
In case of fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed.
In case of a new dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
In case of backwards incompatible changes please leave a note in a newsfragment file, named `{pr_number}.significant.rst` or `{issue_number}.significant.rst`, in [newsfragments](https://github.com/apache/airflow/tree/main/newsfragments).
--
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] Handle backfill for DAGs with when depends_on_past is True [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #45731:
URL: https://github.com/apache/airflow/pull/45731#discussion_r1925118683
##########
airflow/api_fastapi/core_api/routes/public/backfills.py:
##########
@@ -184,10 +187,7 @@ def cancel_backfill(backfill_id, session: SessionDep) -> BackfillResponse:
@backfills_router.post(
path="",
responses=create_openapi_http_exception_doc(
- [
- status.HTTP_404_NOT_FOUND,
- status.HTTP_409_CONFLICT,
- ]
+ [status.HTTP_404_NOT_FOUND, status.HTTP_409_CONFLICT, status.HTTP_422_UNPROCESSABLE_ENTITY]
Review Comment:
Same here
##########
airflow/api_fastapi/core_api/routes/public/backfills.py:
##########
@@ -184,10 +187,7 @@ def cancel_backfill(backfill_id, session: SessionDep) -> BackfillResponse:
@backfills_router.post(
path="",
responses=create_openapi_http_exception_doc(
- [
- status.HTTP_404_NOT_FOUND,
- status.HTTP_409_CONFLICT,
- ]
+ [status.HTTP_404_NOT_FOUND, status.HTTP_409_CONFLICT, status.HTTP_422_UNPROCESSABLE_ENTITY]
Review Comment:
Same here for `status.HTTP_422_UNPROCESSABLE_ENTITY` (to remove)
--
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] Handle backfill for DAGs with when depends_on_past is True [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #45731:
URL: https://github.com/apache/airflow/pull/45731#discussion_r1922255923
##########
tests/api_fastapi/core_api/routes/public/test_backfills.py:
##########
@@ -267,6 +267,53 @@ def test_no_schedule_dag(self, session, dag_maker, test_client):
assert response.status_code == 409
assert response.json().get("detail") == f"{dag.dag_id} has no schedule"
+ @pytest.mark.parametrize(
+ "repro_act, repro_exp,run_backwards, status_code",
+ [
+ ("none", ReprocessBehavior.NONE, False, 409),
+ ("completed", ReprocessBehavior.COMPLETED, False, 200),
+ ("completed", ReprocessBehavior.COMPLETED, True, 409),
+ ],
+ )
+ def test_create_backfill_with_depends_on_past(
+ self, repro_act, repro_exp, run_backwards, status_code, session, dag_maker, test_client
+ ):
+ with dag_maker(session=session, dag_id="TEST_DAG_1", schedule="0 * * * *") as dag:
+ EmptyOperator(task_id="mytask", depends_on_past=True)
+ session.query(DagModel).all()
+ session.commit()
+ from_date = pendulum.parse("2024-01-01")
+ from_date_iso = to_iso(from_date)
+ to_date = pendulum.parse("2024-02-01")
+ to_date_iso = to_iso(to_date)
+ max_active_runs = 5
+ data = {
+ "dag_id": dag.dag_id,
+ "from_date": f"{from_date_iso}",
+ "to_date": f"{to_date_iso}",
+ "max_active_runs": max_active_runs,
+ "run_backwards": run_backwards,
+ "dag_run_conf": {"param1": "val1", "param2": True},
+ "dry_run": False,
+ "reprocess_behavior": repro_act,
+ }
+ response = test_client.post(
+ url="/public/backfills",
+ json=data,
+ )
+ assert response.status_code == status_code
+ if response.status_code == 409 and repro_act == "none":
+ assert (
+ response.json().get("detail")
+ == f"{dag.dag_id} has tasks for which depends_on_past=True. You must set reprocess behavior to reprocess completed or reprocess failed."
+ )
+
+ if response.status_code == 409 and repro_act == "completed":
Review Comment:
If response is not success then handle errors
```suggestion
if response.status_code != 200:
```
```
if run_backward:
...
else:
...
```
To follow the same logic as the code implementation. First `if reverse`, then `if reprocess_behavior ...`. Swapping the assertion order and if condition makes it harder to read and understand if all codepath are taken care of for no real 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] Handle backfill for DAGs with when depends_on_past is True [airflow]
Posted by "vatsrahul1001 (via GitHub)" <gi...@apache.org>.
vatsrahul1001 commented on code in PR #45731:
URL: https://github.com/apache/airflow/pull/45731#discussion_r1922347514
##########
airflow/models/backfill.py:
##########
@@ -75,6 +75,22 @@ class DagNoScheduleException(AirflowException):
"""
+class InvalidBackfillDirection(AirflowException):
+ """
+ Raised when backfill is attempted in reverse with tasks that depend on past runs.
+
+ :meta private:
+ """
+
+
+class InvalidReprocessBehavior(AirflowException):
+ """
+ Raised when reprocess behavior is not set for tasks with depends_on_past=True.
Review Comment:
Done
--
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] Handle backfill for DAGs with when depends_on_past is True [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #45731:
URL: https://github.com/apache/airflow/pull/45731#discussion_r1922579666
##########
tests/api_fastapi/core_api/routes/public/test_backfills.py:
##########
@@ -264,95 +264,191 @@ def test_no_schedule_dag(self, session, dag_maker, test_client):
url="/public/backfills",
json=data,
)
- assert response.status_code == 409
+ assert response.status_code == 422
assert response.json().get("detail") == f"{dag.dag_id} has no schedule"
-
-class TestCreateBackfillDryRun(TestBackfillEndpoint):
@pytest.mark.parametrize(
- "reprocess_behavior, expected_dates",
+ "repro_act, repro_exp, run_backwards, status_code",
[
- (
- "none",
- [
- {"logical_date": "2024-01-01T00:00:00Z"},
- {"logical_date": "2024-01-04T00:00:00Z"},
- {"logical_date": "2024-01-05T00:00:00Z"},
- ],
- ),
- (
- "failed",
- [
- {"logical_date": "2024-01-01T00:00:00Z"},
- {"logical_date": "2024-01-03T00:00:00Z"}, # Reprocess failed
- {"logical_date": "2024-01-04T00:00:00Z"},
- {"logical_date": "2024-01-05T00:00:00Z"},
- ],
- ),
- (
- "completed",
- [
- {"logical_date": "2024-01-01T00:00:00Z"},
- {"logical_date": "2024-01-02T00:00:00Z"}, # Reprocess all
- {"logical_date": "2024-01-03T00:00:00Z"},
- {"logical_date": "2024-01-04T00:00:00Z"},
- {"logical_date": "2024-01-05T00:00:00Z"},
- ],
- ),
+ ("none", ReprocessBehavior.NONE, False, 422),
+ ("completed", ReprocessBehavior.COMPLETED, False, 200),
+ ("completed", ReprocessBehavior.COMPLETED, True, 422),
],
)
- def test_create_backfill_dry_run(
- self, session, dag_maker, test_client, reprocess_behavior, expected_dates
+ def test_create_backfill_with_depends_on_past(
+ self, repro_act, repro_exp, run_backwards, status_code, session, dag_maker, test_client
):
- with dag_maker(
- session=session,
- dag_id="TEST_DAG_2",
- schedule="0 0 * * *",
- start_date=pendulum.parse("2024-01-01"),
- ) as dag:
- EmptyOperator(task_id="mytask")
-
+ with dag_maker(session=session, dag_id="TEST_DAG_1", schedule="0 * * * *") as dag:
+ EmptyOperator(task_id="mytask", depends_on_past=True)
+ session.query(DagModel).all()
session.commit()
+ from_date = pendulum.parse("2024-01-01")
+ from_date_iso = to_iso(from_date)
+ to_date = pendulum.parse("2024-02-01")
+ to_date_iso = to_iso(to_date)
+ max_active_runs = 5
+ data = {
+ "dag_id": dag.dag_id,
+ "from_date": f"{from_date_iso}",
+ "to_date": f"{to_date_iso}",
+ "max_active_runs": max_active_runs,
+ "run_backwards": run_backwards,
+ "dag_run_conf": {"param1": "val1", "param2": True},
+ "dry_run": False,
+ "reprocess_behavior": repro_act,
+ }
+ response = test_client.post(
+ url="/public/backfills",
+ json=data,
+ )
+ assert response.status_code == status_code
- existing_dagruns = [
- {"logical_date": pendulum.parse("2024-01-02"), "state": DagRunState.SUCCESS}, # Completed dag run
- {"logical_date": pendulum.parse("2024-01-03"), "state": DagRunState.FAILED}, # Failed dag run
- ]
- for dagrun in existing_dagruns:
- session.add(
- DagRun(
- dag_id=dag.dag_id,
- run_id=f"manual__{dagrun['logical_date'].isoformat()}",
- logical_date=dagrun["logical_date"],
- state=dagrun["state"],
- run_type="scheduled",
+ if response.status_code != 200:
+ if run_backwards:
+ assert (
+ response.json().get("detail")
+ == "Backfill cannot be run in reverse when the DAG has tasks where depends_on_past=True."
+ )
+ else:
+ assert (
+ response.json().get("detail")
+ == f"{dag.dag_id} has tasks for which depends_on_past=True. You must set reprocess behavior to reprocess completed or reprocess failed."
)
- )
- session.commit()
+
+class TestCreateBackfillDryRun(TestBackfillEndpoint):
+ # @pytest.mark.parametrize(
Review Comment:
Hi @vatsrahul1001, is the commented-out test case here intentionally meant to stay, or should it be removed?
--
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] Handle backfill for DAGs with when depends_on_past is True [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #45731:
URL: https://github.com/apache/airflow/pull/45731#discussion_r1924048743
##########
airflow/models/backfill.py:
##########
@@ -75,6 +75,22 @@ class DagNoScheduleException(AirflowException):
"""
+class InvalidBackfillDirection(AirflowException):
+ """
+ Raised when backfill is attempted in reverse order with tasks that depend on past runs.
+
+ :meta private:
+ """
+
+
+class InvalidReprocessBehavior(AirflowException):
+ """
+ Raised when a backfill cannot be completed because the reprocess behavior is not valid.
+
+ :meta private:
+ """
Review Comment:
Here are some examples of the corresponding changes to exceptions for the router refactor. Additionally, the `__str__` method has been added, which could be helpful for debugging during traceback.
```suggestion
class InvalidReprocessBehavior(AirflowException):
"""
Raised when a backfill cannot be completed because the reprocess behavior is not valid.
:meta private:
"""
def __init__(self, dag_id: str):
self.dag_id = dag_id
self.message = (
f"{dag_id} has tasks for which depends_on_past=True. "
"You must set reprocess behavior to reprocess completed or reprocess failed."
)
super().__init__(self.message)
def __str__(self):
return self.message
```
##########
airflow/api_fastapi/core_api/routes/public/backfills.py:
##########
@@ -213,9 +212,21 @@ def create_backfill(
)
except DagNoScheduleException:
Review Comment:
I think we can consolidate all the 422 Unprocessable Entity errors here and extract the original error from `e.message`. WDYT?
```suggestion
except (
AlreadyRunningBackfill,
DagNoScheduleException,
InvalidReprocessBehavior,
InvalidBackfillDirection,
) as e:
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail=e.message,
)
```
--
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] Handle backfill for DAGs with when depends_on_past is True [airflow]
Posted by "vatsrahul1001 (via GitHub)" <gi...@apache.org>.
vatsrahul1001 commented on code in PR #45731:
URL: https://github.com/apache/airflow/pull/45731#discussion_r1922617451
##########
tests/api_fastapi/core_api/routes/public/test_backfills.py:
##########
@@ -264,95 +264,191 @@ def test_no_schedule_dag(self, session, dag_maker, test_client):
url="/public/backfills",
json=data,
)
- assert response.status_code == 409
+ assert response.status_code == 422
assert response.json().get("detail") == f"{dag.dag_id} has no schedule"
-
-class TestCreateBackfillDryRun(TestBackfillEndpoint):
@pytest.mark.parametrize(
- "reprocess_behavior, expected_dates",
+ "repro_act, repro_exp, run_backwards, status_code",
[
- (
- "none",
- [
- {"logical_date": "2024-01-01T00:00:00Z"},
- {"logical_date": "2024-01-04T00:00:00Z"},
- {"logical_date": "2024-01-05T00:00:00Z"},
- ],
- ),
- (
- "failed",
- [
- {"logical_date": "2024-01-01T00:00:00Z"},
- {"logical_date": "2024-01-03T00:00:00Z"}, # Reprocess failed
- {"logical_date": "2024-01-04T00:00:00Z"},
- {"logical_date": "2024-01-05T00:00:00Z"},
- ],
- ),
- (
- "completed",
- [
- {"logical_date": "2024-01-01T00:00:00Z"},
- {"logical_date": "2024-01-02T00:00:00Z"}, # Reprocess all
- {"logical_date": "2024-01-03T00:00:00Z"},
- {"logical_date": "2024-01-04T00:00:00Z"},
- {"logical_date": "2024-01-05T00:00:00Z"},
- ],
- ),
+ ("none", ReprocessBehavior.NONE, False, 422),
+ ("completed", ReprocessBehavior.COMPLETED, False, 200),
+ ("completed", ReprocessBehavior.COMPLETED, True, 422),
],
)
- def test_create_backfill_dry_run(
- self, session, dag_maker, test_client, reprocess_behavior, expected_dates
+ def test_create_backfill_with_depends_on_past(
+ self, repro_act, repro_exp, run_backwards, status_code, session, dag_maker, test_client
):
- with dag_maker(
- session=session,
- dag_id="TEST_DAG_2",
- schedule="0 0 * * *",
- start_date=pendulum.parse("2024-01-01"),
- ) as dag:
- EmptyOperator(task_id="mytask")
-
+ with dag_maker(session=session, dag_id="TEST_DAG_1", schedule="0 * * * *") as dag:
+ EmptyOperator(task_id="mytask", depends_on_past=True)
+ session.query(DagModel).all()
session.commit()
+ from_date = pendulum.parse("2024-01-01")
+ from_date_iso = to_iso(from_date)
+ to_date = pendulum.parse("2024-02-01")
+ to_date_iso = to_iso(to_date)
+ max_active_runs = 5
+ data = {
+ "dag_id": dag.dag_id,
+ "from_date": f"{from_date_iso}",
+ "to_date": f"{to_date_iso}",
+ "max_active_runs": max_active_runs,
+ "run_backwards": run_backwards,
+ "dag_run_conf": {"param1": "val1", "param2": True},
+ "dry_run": False,
+ "reprocess_behavior": repro_act,
+ }
+ response = test_client.post(
+ url="/public/backfills",
+ json=data,
+ )
+ assert response.status_code == status_code
- existing_dagruns = [
- {"logical_date": pendulum.parse("2024-01-02"), "state": DagRunState.SUCCESS}, # Completed dag run
- {"logical_date": pendulum.parse("2024-01-03"), "state": DagRunState.FAILED}, # Failed dag run
- ]
- for dagrun in existing_dagruns:
- session.add(
- DagRun(
- dag_id=dag.dag_id,
- run_id=f"manual__{dagrun['logical_date'].isoformat()}",
- logical_date=dagrun["logical_date"],
- state=dagrun["state"],
- run_type="scheduled",
+ if response.status_code != 200:
+ if run_backwards:
+ assert (
+ response.json().get("detail")
+ == "Backfill cannot be run in reverse when the DAG has tasks where depends_on_past=True."
+ )
+ else:
+ assert (
+ response.json().get("detail")
+ == f"{dag.dag_id} has tasks for which depends_on_past=True. You must set reprocess behavior to reprocess completed or reprocess failed."
)
- )
- session.commit()
+
+class TestCreateBackfillDryRun(TestBackfillEndpoint):
+ # @pytest.mark.parametrize(
Review Comment:
@jason810496 commented by mistake, Fixed 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] Handle backfill for DAGs with when depends_on_past is True [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #45731:
URL: https://github.com/apache/airflow/pull/45731#discussion_r1924092898
##########
airflow/api_fastapi/core_api/routes/public/backfills.py:
##########
@@ -259,6 +267,18 @@ def create_backfill_dry_run(
)
except DagNoScheduleException:
raise HTTPException(
- status_code=status.HTTP_409_CONFLICT,
+ status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail=f"{body.dag_id} has no schedule",
)
+
+ except InvalidReprocessBehavior:
+ raise HTTPException(
+ status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+ detail=f"{body.dag_id} has tasks for which depends_on_past=True. "
+ "You must set reprocess behavior to reprocess completed or reprocess failed.",
+ )
+ except InvalidBackfillDirection:
+ raise HTTPException(
+ status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+ detail="Backfill cannot be run in reverse when the DAG has tasks where depends_on_past=True.",
+ )
Review Comment:
I know why you couldn't retrieve the error message.
You can do that instead:
```suggestion
except (InvalidReprocessBehavior, InvalidBackfillDirection, DagNoScheduleException) as e:
raise RequestValidationError(str(e))
```
(FastAPI catches those and throws 422 automtaically for us) (Same for the other endpoit)
> Note that DagNoScheduleException is also handled 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] Handle backfill for DAGs with when depends_on_past is True [airflow]
Posted by "vatsrahul1001 (via GitHub)" <gi...@apache.org>.
vatsrahul1001 commented on code in PR #45731:
URL: https://github.com/apache/airflow/pull/45731#discussion_r1922350800
##########
tests/api_fastapi/core_api/routes/public/test_backfills.py:
##########
@@ -267,6 +267,53 @@ def test_no_schedule_dag(self, session, dag_maker, test_client):
assert response.status_code == 409
assert response.json().get("detail") == f"{dag.dag_id} has no schedule"
+ @pytest.mark.parametrize(
+ "repro_act, repro_exp,run_backwards, status_code",
Review Comment:
done
##########
tests/api_fastapi/core_api/routes/public/test_backfills.py:
##########
@@ -267,6 +267,53 @@ def test_no_schedule_dag(self, session, dag_maker, test_client):
assert response.status_code == 409
assert response.json().get("detail") == f"{dag.dag_id} has no schedule"
+ @pytest.mark.parametrize(
+ "repro_act, repro_exp,run_backwards, status_code",
+ [
+ ("none", ReprocessBehavior.NONE, False, 409),
+ ("completed", ReprocessBehavior.COMPLETED, False, 200),
+ ("completed", ReprocessBehavior.COMPLETED, True, 409),
+ ],
+ )
+ def test_create_backfill_with_depends_on_past(
Review Comment:
done
--
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 Handle backfill for DAGs with when depends_on_past is True [airflow]
Posted by "vatsrahul1001 (via GitHub)" <gi...@apache.org>.
vatsrahul1001 commented on code in PR #45731:
URL: https://github.com/apache/airflow/pull/45731#discussion_r1925265578
##########
airflow/api_fastapi/core_api/routes/public/backfills.py:
##########
@@ -206,31 +206,20 @@ def create_backfill(
reprocess_behavior=backfill_request.reprocess_behavior,
)
return BackfillResponse.model_validate(backfill_obj)
- except AlreadyRunningBackfill:
- raise HTTPException(
- status_code=status.HTTP_409_CONFLICT,
- detail=f"There is already a running backfill for dag {backfill_request.dag_id}",
- )
- except DagNoScheduleException:
- raise HTTPException(
- status_code=status.HTTP_409_CONFLICT,
- detail=f"{backfill_request.dag_id} has no schedule",
- )
-
- except DagNotFound:
- raise HTTPException(
- status_code=status.HTTP_404_NOT_FOUND,
- detail=f"Could not find dag {backfill_request.dag_id}",
- )
+ except (
+ AlreadyRunningBackfill,
+ InvalidReprocessBehavior,
+ InvalidBackfillDirection,
+ DagNoScheduleException,
+ DagNotFound,
+ ) as e:
+ raise RequestValidationError(str(e))
@backfills_router.post(
path="/dry_run",
responses=create_openapi_http_exception_doc(
- [
- status.HTTP_404_NOT_FOUND,
- status.HTTP_409_CONFLICT,
- ]
+ [status.HTTP_404_NOT_FOUND, status.HTTP_409_CONFLICT, status.HTTP_422_UNPROCESSABLE_ENTITY]
Review Comment:
Fixed
--
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] Handle backfill for DAGs with when depends_on_past is True [airflow]
Posted by "vatsrahul1001 (via GitHub)" <gi...@apache.org>.
vatsrahul1001 commented on code in PR #45731:
URL: https://github.com/apache/airflow/pull/45731#discussion_r1924720595
##########
airflow/api_fastapi/core_api/routes/public/backfills.py:
##########
@@ -259,6 +267,18 @@ def create_backfill_dry_run(
)
except DagNoScheduleException:
raise HTTPException(
- status_code=status.HTTP_409_CONFLICT,
+ status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail=f"{body.dag_id} has no schedule",
)
+
+ except InvalidReprocessBehavior:
+ raise HTTPException(
+ status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+ detail=f"{body.dag_id} has tasks for which depends_on_past=True. "
+ "You must set reprocess behavior to reprocess completed or reprocess failed.",
+ )
+ except InvalidBackfillDirection:
+ raise HTTPException(
+ status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+ detail="Backfill cannot be run in reverse when the DAG has tasks where depends_on_past=True.",
+ )
Review Comment:
This works, thanks @pierrejeambrun implemented 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] Handle backfill for DAGs with when depends_on_past is True [airflow]
Posted by "vatsrahul1001 (via GitHub)" <gi...@apache.org>.
vatsrahul1001 commented on code in PR #45731:
URL: https://github.com/apache/airflow/pull/45731#discussion_r1922641257
##########
airflow/api_fastapi/core_api/routes/public/backfills.py:
##########
@@ -216,6 +218,17 @@ def create_backfill(
status_code=status.HTTP_409_CONFLICT,
detail=f"{backfill_request.dag_id} has no schedule",
)
+ except InvalidReprocessBehavior:
+ raise HTTPException(
+ status_code=status.HTTP_409_CONFLICT,
+ detail=f"{backfill_request.dag_id} has tasks for which depends_on_past=True. "
+ "You must set reprocess behavior to reprocess completed or reprocess failed.",
+ )
+ except InvalidBackfillDirection:
+ raise HTTPException(
+ status_code=status.HTTP_409_CONFLICT,
+ detail="Backfill cannot be run in reverse when the DAG has tasks where depends_on_past=True.",
+ )
Review Comment:
No. I tried adding this to class
```
class InvalidReprocessBehavior(AirflowException):
"""
Raised when a backfill cannot be completed because the reprocess behavior is not valid.
:meta private:
"""
def __init__(self, dag_id: str):
self.dag_id = dag_id
self.message = (
f"{dag_id} has tasks for which depends_on_past=True. "
"You must set reprocess behavior to reprocess completed or reprocess failed."
)
super().__init__(self.message)
```
However, I am only getting below error message
```
{
"detail": "Unprocessable Entity"
}
```
--
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] Handle backfill for DAGs with when depends_on_past is True [airflow]
Posted by "vatsrahul1001 (via GitHub)" <gi...@apache.org>.
vatsrahul1001 commented on code in PR #45731:
URL: https://github.com/apache/airflow/pull/45731#discussion_r1924720912
##########
airflow/api_fastapi/core_api/routes/public/backfills.py:
##########
@@ -213,9 +212,21 @@ def create_backfill(
)
except DagNoScheduleException:
Review Comment:
Implemented this https://github.com/apache/airflow/pull/45731#discussion_r1924092898
--
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] Handle backfill for DAGs with when depends_on_past is True [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #45731:
URL: https://github.com/apache/airflow/pull/45731#discussion_r1922255923
##########
tests/api_fastapi/core_api/routes/public/test_backfills.py:
##########
@@ -267,6 +267,53 @@ def test_no_schedule_dag(self, session, dag_maker, test_client):
assert response.status_code == 409
assert response.json().get("detail") == f"{dag.dag_id} has no schedule"
+ @pytest.mark.parametrize(
+ "repro_act, repro_exp,run_backwards, status_code",
+ [
+ ("none", ReprocessBehavior.NONE, False, 409),
+ ("completed", ReprocessBehavior.COMPLETED, False, 200),
+ ("completed", ReprocessBehavior.COMPLETED, True, 409),
+ ],
+ )
+ def test_create_backfill_with_depends_on_past(
+ self, repro_act, repro_exp, run_backwards, status_code, session, dag_maker, test_client
+ ):
+ with dag_maker(session=session, dag_id="TEST_DAG_1", schedule="0 * * * *") as dag:
+ EmptyOperator(task_id="mytask", depends_on_past=True)
+ session.query(DagModel).all()
+ session.commit()
+ from_date = pendulum.parse("2024-01-01")
+ from_date_iso = to_iso(from_date)
+ to_date = pendulum.parse("2024-02-01")
+ to_date_iso = to_iso(to_date)
+ max_active_runs = 5
+ data = {
+ "dag_id": dag.dag_id,
+ "from_date": f"{from_date_iso}",
+ "to_date": f"{to_date_iso}",
+ "max_active_runs": max_active_runs,
+ "run_backwards": run_backwards,
+ "dag_run_conf": {"param1": "val1", "param2": True},
+ "dry_run": False,
+ "reprocess_behavior": repro_act,
+ }
+ response = test_client.post(
+ url="/public/backfills",
+ json=data,
+ )
+ assert response.status_code == status_code
+ if response.status_code == 409 and repro_act == "none":
+ assert (
+ response.json().get("detail")
+ == f"{dag.dag_id} has tasks for which depends_on_past=True. You must set reprocess behavior to reprocess completed or reprocess failed."
+ )
+
+ if response.status_code == 409 and repro_act == "completed":
Review Comment:
If response is not success then handle errors
```suggestion
if response.status_code != 200:
```
```
if run_backward:
...
else:
...
```
To follow the same logic as the code implementation. First `if reverse`, then `if reprocess_behavior ...`. Swapping the assertion order and if conditions makes it harder to read and understand if all codepath are taken care of, for no real 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] Handle backfill for DAGs with when depends_on_past is True [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on PR #45731:
URL: https://github.com/apache/airflow/pull/45731#issuecomment-2602175560
cc: @dstandish
--
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] Handle backfill for DAGs with when depends_on_past is True [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #45731:
URL: https://github.com/apache/airflow/pull/45731#discussion_r1922612741
##########
airflow/api_fastapi/core_api/routes/public/backfills.py:
##########
@@ -216,6 +218,17 @@ def create_backfill(
status_code=status.HTTP_409_CONFLICT,
detail=f"{backfill_request.dag_id} has no schedule",
)
+ except InvalidReprocessBehavior:
+ raise HTTPException(
+ status_code=status.HTTP_409_CONFLICT,
+ detail=f"{backfill_request.dag_id} has tasks for which depends_on_past=True. "
+ "You must set reprocess behavior to reprocess completed or reprocess failed.",
+ )
+ except InvalidBackfillDirection:
+ raise HTTPException(
+ status_code=status.HTTP_409_CONFLICT,
+ detail="Backfill cannot be run in reverse when the DAG has tasks where depends_on_past=True.",
+ )
Review Comment:
> somehow exception message was not coming from InvalidReprocessBehavior and InvalidBackfillDirection
Were you able to solve this? You should be able to get the error message raised from the backfill 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] Handle backfill for DAGs with when depends_on_past is True [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on PR #45731:
URL: https://github.com/apache/airflow/pull/45731#issuecomment-2602795515
Don't hesitate to `resolve` comments that you addressed. Also I think a couple of them are marked as `done` but I don't see the related change yet. (maybe you have some local unpushed 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] Handle backfill for DAGs with when depends_on_past is True [airflow]
Posted by "vatsrahul1001 (via GitHub)" <gi...@apache.org>.
vatsrahul1001 commented on code in PR #45731:
URL: https://github.com/apache/airflow/pull/45731#discussion_r1922351962
##########
tests/api_fastapi/core_api/routes/public/test_backfills.py:
##########
@@ -267,6 +267,53 @@ def test_no_schedule_dag(self, session, dag_maker, test_client):
assert response.status_code == 409
assert response.json().get("detail") == f"{dag.dag_id} has no schedule"
+ @pytest.mark.parametrize(
+ "repro_act, repro_exp,run_backwards, status_code",
+ [
+ ("none", ReprocessBehavior.NONE, False, 409),
+ ("completed", ReprocessBehavior.COMPLETED, False, 200),
+ ("completed", ReprocessBehavior.COMPLETED, True, 409),
+ ],
+ )
+ def test_create_backfill_with_depends_on_past(
+ self, repro_act, repro_exp, run_backwards, status_code, session, dag_maker, test_client
+ ):
+ with dag_maker(session=session, dag_id="TEST_DAG_1", schedule="0 * * * *") as dag:
+ EmptyOperator(task_id="mytask", depends_on_past=True)
+ session.query(DagModel).all()
+ session.commit()
Review Comment:
It add `TEST_DAG_1` to `DagModel`
##########
tests/api_fastapi/core_api/routes/public/test_backfills.py:
##########
@@ -267,6 +267,53 @@ def test_no_schedule_dag(self, session, dag_maker, test_client):
assert response.status_code == 409
assert response.json().get("detail") == f"{dag.dag_id} has no schedule"
+ @pytest.mark.parametrize(
+ "repro_act, repro_exp,run_backwards, status_code",
+ [
+ ("none", ReprocessBehavior.NONE, False, 409),
+ ("completed", ReprocessBehavior.COMPLETED, False, 200),
+ ("completed", ReprocessBehavior.COMPLETED, True, 409),
+ ],
+ )
+ def test_create_backfill_with_depends_on_past(
+ self, repro_act, repro_exp, run_backwards, status_code, session, dag_maker, test_client
+ ):
+ with dag_maker(session=session, dag_id="TEST_DAG_1", schedule="0 * * * *") as dag:
+ EmptyOperator(task_id="mytask", depends_on_past=True)
+ session.query(DagModel).all()
+ session.commit()
+ from_date = pendulum.parse("2024-01-01")
+ from_date_iso = to_iso(from_date)
+ to_date = pendulum.parse("2024-02-01")
+ to_date_iso = to_iso(to_date)
+ max_active_runs = 5
+ data = {
+ "dag_id": dag.dag_id,
+ "from_date": f"{from_date_iso}",
+ "to_date": f"{to_date_iso}",
+ "max_active_runs": max_active_runs,
+ "run_backwards": run_backwards,
+ "dag_run_conf": {"param1": "val1", "param2": True},
+ "dry_run": False,
+ "reprocess_behavior": repro_act,
+ }
+ response = test_client.post(
+ url="/public/backfills",
+ json=data,
+ )
+ assert response.status_code == status_code
+ if response.status_code == 409 and repro_act == "none":
+ assert (
+ response.json().get("detail")
+ == f"{dag.dag_id} has tasks for which depends_on_past=True. You must set reprocess behavior to reprocess completed or reprocess failed."
+ )
+
+ if response.status_code == 409 and repro_act == "completed":
Review Comment:
done
--
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] Handle backfill for DAGs with when depends_on_past is True [airflow]
Posted by "vatsrahul1001 (via GitHub)" <gi...@apache.org>.
vatsrahul1001 commented on code in PR #45731:
URL: https://github.com/apache/airflow/pull/45731#discussion_r1922351962
##########
tests/api_fastapi/core_api/routes/public/test_backfills.py:
##########
@@ -267,6 +267,53 @@ def test_no_schedule_dag(self, session, dag_maker, test_client):
assert response.status_code == 409
assert response.json().get("detail") == f"{dag.dag_id} has no schedule"
+ @pytest.mark.parametrize(
+ "repro_act, repro_exp,run_backwards, status_code",
+ [
+ ("none", ReprocessBehavior.NONE, False, 409),
+ ("completed", ReprocessBehavior.COMPLETED, False, 200),
+ ("completed", ReprocessBehavior.COMPLETED, True, 409),
+ ],
+ )
+ def test_create_backfill_with_depends_on_past(
+ self, repro_act, repro_exp, run_backwards, status_code, session, dag_maker, test_client
+ ):
+ with dag_maker(session=session, dag_id="TEST_DAG_1", schedule="0 * * * *") as dag:
+ EmptyOperator(task_id="mytask", depends_on_past=True)
+ session.query(DagModel).all()
+ session.commit()
Review Comment:
It add's `TEST_DAG_1` to `DagModel`
--
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] Handle backfill for DAGs with when depends_on_past is True [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #45731:
URL: https://github.com/apache/airflow/pull/45731#discussion_r1925081494
##########
airflow/api_fastapi/core_api/routes/public/backfills.py:
##########
@@ -206,31 +206,20 @@ def create_backfill(
reprocess_behavior=backfill_request.reprocess_behavior,
)
return BackfillResponse.model_validate(backfill_obj)
- except AlreadyRunningBackfill:
- raise HTTPException(
- status_code=status.HTTP_409_CONFLICT,
- detail=f"There is already a running backfill for dag {backfill_request.dag_id}",
- )
- except DagNoScheduleException:
- raise HTTPException(
- status_code=status.HTTP_409_CONFLICT,
- detail=f"{backfill_request.dag_id} has no schedule",
- )
-
- except DagNotFound:
- raise HTTPException(
- status_code=status.HTTP_404_NOT_FOUND,
- detail=f"Could not find dag {backfill_request.dag_id}",
- )
+ except (
+ AlreadyRunningBackfill,
+ InvalidReprocessBehavior,
+ InvalidBackfillDirection,
+ DagNoScheduleException,
+ DagNotFound,
+ ) as e:
+ raise RequestValidationError(str(e))
@backfills_router.post(
path="/dry_run",
responses=create_openapi_http_exception_doc(
- [
- status.HTTP_404_NOT_FOUND,
- status.HTTP_409_CONFLICT,
- ]
+ [status.HTTP_404_NOT_FOUND, status.HTTP_409_CONFLICT, status.HTTP_422_UNPROCESSABLE_ENTITY]
Review Comment:
`HTTP_422_UNPROCESSABLE_ENTITY` this is already handled by pydantic natively, we don't need to specify it.
##########
airflow/api_fastapi/core_api/routes/public/backfills.py:
##########
@@ -252,13 +241,6 @@ def create_backfill_dry_run(
backfills = [DryRunBackfillResponse(logical_date=d) for d in backfills_dry_run]
return DryRunBackfillCollectionResponse(backfills=backfills, total_entries=len(backfills_dry_run))
- except DagNotFound:
- raise HTTPException(
- status_code=status.HTTP_404_NOT_FOUND,
- detail=f"Could not find dag {body.dag_id}",
- )
- except DagNoScheduleException:
- raise HTTPException(
- status_code=status.HTTP_409_CONFLICT,
- detail=f"{body.dag_id} has no schedule",
- )
+
+ except (InvalidReprocessBehavior, InvalidBackfillDirection, DagNoScheduleException, DagNotFound) as e:
Review Comment:
`DagNotFound` keep old handling.
##########
airflow/api_fastapi/core_api/routes/public/backfills.py:
##########
@@ -206,31 +206,20 @@ def create_backfill(
reprocess_behavior=backfill_request.reprocess_behavior,
)
return BackfillResponse.model_validate(backfill_obj)
- except AlreadyRunningBackfill:
- raise HTTPException(
- status_code=status.HTTP_409_CONFLICT,
- detail=f"There is already a running backfill for dag {backfill_request.dag_id}",
- )
- except DagNoScheduleException:
- raise HTTPException(
- status_code=status.HTTP_409_CONFLICT,
- detail=f"{backfill_request.dag_id} has no schedule",
- )
-
- except DagNotFound:
- raise HTTPException(
- status_code=status.HTTP_404_NOT_FOUND,
- detail=f"Could not find dag {backfill_request.dag_id}",
- )
+ except (
+ AlreadyRunningBackfill,
Review Comment:
DagNotFound is a 404, same we don't want to return a validation error 422 on that.
##########
airflow/api_fastapi/core_api/routes/public/backfills.py:
##########
@@ -206,31 +206,20 @@ def create_backfill(
reprocess_behavior=backfill_request.reprocess_behavior,
)
return BackfillResponse.model_validate(backfill_obj)
- except AlreadyRunningBackfill:
- raise HTTPException(
- status_code=status.HTTP_409_CONFLICT,
- detail=f"There is already a running backfill for dag {backfill_request.dag_id}",
- )
- except DagNoScheduleException:
- raise HTTPException(
- status_code=status.HTTP_409_CONFLICT,
- detail=f"{backfill_request.dag_id} has no schedule",
- )
-
- except DagNotFound:
- raise HTTPException(
- status_code=status.HTTP_404_NOT_FOUND,
- detail=f"Could not find dag {backfill_request.dag_id}",
- )
+ except (
+ AlreadyRunningBackfill,
Review Comment:
`AlreadyRunningBackfill` needs to remain on it's own because it's a `409` conflict.
`RequestValidationError` will return a 422.
--
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] Handle backfill for DAGs with when depends_on_past is True [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun commented on code in PR #45731:
URL: https://github.com/apache/airflow/pull/45731#discussion_r1922244421
##########
airflow/models/backfill.py:
##########
@@ -75,6 +75,22 @@ class DagNoScheduleException(AirflowException):
"""
+class InvalidBackfillDirection(AirflowException):
+ """
+ Raised when backfill is attempted in reverse with tasks that depend on past runs.
Review Comment:
```suggestion
Raised when backfill is attempted in reverse order with tasks that depend on past runs.
```
##########
tests/api_fastapi/core_api/routes/public/test_backfills.py:
##########
@@ -267,6 +267,53 @@ def test_no_schedule_dag(self, session, dag_maker, test_client):
assert response.status_code == 409
assert response.json().get("detail") == f"{dag.dag_id} has no schedule"
+ @pytest.mark.parametrize(
+ "repro_act, repro_exp,run_backwards, status_code",
Review Comment:
```suggestion
"repro_act, repro_exp, run_backwards, status_code",
```
##########
airflow/api_fastapi/core_api/routes/public/backfills.py:
##########
@@ -216,6 +218,17 @@ def create_backfill(
status_code=status.HTTP_409_CONFLICT,
detail=f"{backfill_request.dag_id} has no schedule",
)
+ except InvalidReprocessBehavior:
+ raise HTTPException(
+ status_code=status.HTTP_409_CONFLICT,
+ detail=f"{backfill_request.dag_id} has tasks for which depends_on_past=True. "
+ "You must set reprocess behavior to reprocess completed or reprocess failed.",
+ )
+ except InvalidBackfillDirection:
+ raise HTTPException(
+ status_code=status.HTTP_409_CONFLICT,
+ detail="Backfill cannot be run in reverse when the DAG has tasks where depends_on_past=True.",
+ )
Review Comment:
Same for DagNoScheduleException, 409 does not seem to be the most appropriate
##########
tests/api_fastapi/core_api/routes/public/test_backfills.py:
##########
@@ -267,6 +267,53 @@ def test_no_schedule_dag(self, session, dag_maker, test_client):
assert response.status_code == 409
assert response.json().get("detail") == f"{dag.dag_id} has no schedule"
+ @pytest.mark.parametrize(
+ "repro_act, repro_exp,run_backwards, status_code",
+ [
+ ("none", ReprocessBehavior.NONE, False, 409),
+ ("completed", ReprocessBehavior.COMPLETED, False, 200),
+ ("completed", ReprocessBehavior.COMPLETED, True, 409),
+ ],
+ )
+ def test_create_backfill_with_depends_on_past(
+ self, repro_act, repro_exp, run_backwards, status_code, session, dag_maker, test_client
+ ):
+ with dag_maker(session=session, dag_id="TEST_DAG_1", schedule="0 * * * *") as dag:
+ EmptyOperator(task_id="mytask", depends_on_past=True)
+ session.query(DagModel).all()
+ session.commit()
Review Comment:
What are those two lines for ? Query without using any results, and committing an empty session I believe.
##########
airflow/models/backfill.py:
##########
@@ -75,6 +75,22 @@ class DagNoScheduleException(AirflowException):
"""
+class InvalidBackfillDirection(AirflowException):
+ """
+ Raised when backfill is attempted in reverse with tasks that depend on past runs.
+
+ :meta private:
+ """
+
+
+class InvalidReprocessBehavior(AirflowException):
+ """
+ Raised when reprocess behavior is not set for tasks with depends_on_past=True.
Review Comment:
```suggestion
class InvalidReprocessBehavior(AirflowException):
"""
Raised when a backfill cannot be completed because the reprocess behavior is not valid.
```
##########
airflow/api_fastapi/core_api/routes/public/backfills.py:
##########
@@ -216,6 +218,17 @@ def create_backfill(
status_code=status.HTTP_409_CONFLICT,
detail=f"{backfill_request.dag_id} has no schedule",
)
+ except InvalidReprocessBehavior:
+ raise HTTPException(
+ status_code=status.HTTP_409_CONFLICT,
+ detail=f"{backfill_request.dag_id} has tasks for which depends_on_past=True. "
+ "You must set reprocess behavior to reprocess completed or reprocess failed.",
+ )
+ except InvalidBackfillDirection:
+ raise HTTPException(
+ status_code=status.HTTP_409_CONFLICT,
+ detail="Backfill cannot be run in reverse when the DAG has tasks where depends_on_past=True.",
+ )
Review Comment:
I'm not sure that those are `409` conflict. Maybe more of a 422 unprocessable entity. I think the default error message from the `InvalidReprocessBehavior` and `InvalidBackfillDirection` is good enough and should be retrieved directly from the exception.
##########
tests/api_fastapi/core_api/routes/public/test_backfills.py:
##########
@@ -267,6 +267,53 @@ def test_no_schedule_dag(self, session, dag_maker, test_client):
assert response.status_code == 409
assert response.json().get("detail") == f"{dag.dag_id} has no schedule"
+ @pytest.mark.parametrize(
+ "repro_act, repro_exp,run_backwards, status_code",
+ [
+ ("none", ReprocessBehavior.NONE, False, 409),
+ ("completed", ReprocessBehavior.COMPLETED, False, 200),
+ ("completed", ReprocessBehavior.COMPLETED, True, 409),
+ ],
+ )
+ def test_create_backfill_with_depends_on_past(
Review Comment:
Ideally we would want a similar test for the dry_run endpoint.
##########
tests/api_fastapi/core_api/routes/public/test_backfills.py:
##########
@@ -267,6 +267,53 @@ def test_no_schedule_dag(self, session, dag_maker, test_client):
assert response.status_code == 409
assert response.json().get("detail") == f"{dag.dag_id} has no schedule"
+ @pytest.mark.parametrize(
+ "repro_act, repro_exp,run_backwards, status_code",
+ [
+ ("none", ReprocessBehavior.NONE, False, 409),
+ ("completed", ReprocessBehavior.COMPLETED, False, 200),
+ ("completed", ReprocessBehavior.COMPLETED, True, 409),
+ ],
+ )
+ def test_create_backfill_with_depends_on_past(
+ self, repro_act, repro_exp, run_backwards, status_code, session, dag_maker, test_client
+ ):
+ with dag_maker(session=session, dag_id="TEST_DAG_1", schedule="0 * * * *") as dag:
+ EmptyOperator(task_id="mytask", depends_on_past=True)
+ session.query(DagModel).all()
+ session.commit()
+ from_date = pendulum.parse("2024-01-01")
+ from_date_iso = to_iso(from_date)
+ to_date = pendulum.parse("2024-02-01")
+ to_date_iso = to_iso(to_date)
+ max_active_runs = 5
+ data = {
+ "dag_id": dag.dag_id,
+ "from_date": f"{from_date_iso}",
+ "to_date": f"{to_date_iso}",
+ "max_active_runs": max_active_runs,
+ "run_backwards": run_backwards,
+ "dag_run_conf": {"param1": "val1", "param2": True},
+ "dry_run": False,
+ "reprocess_behavior": repro_act,
+ }
+ response = test_client.post(
+ url="/public/backfills",
+ json=data,
+ )
+ assert response.status_code == status_code
+ if response.status_code == 409 and repro_act == "none":
+ assert (
+ response.json().get("detail")
+ == f"{dag.dag_id} has tasks for which depends_on_past=True. You must set reprocess behavior to reprocess completed or reprocess failed."
+ )
+
+ if response.status_code == 409 and repro_act == "completed":
Review Comment:
If response is not success then handle errors
```suggestion
if response.status_code != 200:
```
```
if run_backward:
...
else:
...
```
To follow the same logic as the code implementation. First if reverse, then if reprocess_behavior. Swapping the assertion order and if condition makes it harder to read and understand if all codepath are taken care of for no real 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] Handle backfill for DAGs with when depends_on_past is True [airflow]
Posted by "vatsrahul1001 (via GitHub)" <gi...@apache.org>.
vatsrahul1001 commented on code in PR #45731:
URL: https://github.com/apache/airflow/pull/45731#discussion_r1922347209
##########
airflow/api_fastapi/core_api/routes/public/backfills.py:
##########
@@ -216,6 +218,17 @@ def create_backfill(
status_code=status.HTTP_409_CONFLICT,
detail=f"{backfill_request.dag_id} has no schedule",
)
+ except InvalidReprocessBehavior:
+ raise HTTPException(
+ status_code=status.HTTP_409_CONFLICT,
+ detail=f"{backfill_request.dag_id} has tasks for which depends_on_past=True. "
+ "You must set reprocess behavior to reprocess completed or reprocess failed.",
+ )
+ except InvalidBackfillDirection:
+ raise HTTPException(
+ status_code=status.HTTP_409_CONFLICT,
+ detail="Backfill cannot be run in reverse when the DAG has tasks where depends_on_past=True.",
+ )
Review Comment:
Implemented 422, tried removing detail from here, somehow exception message was not coming from `InvalidReprocessBehavior` and `InvalidBackfillDirection`
##########
airflow/models/backfill.py:
##########
@@ -75,6 +75,22 @@ class DagNoScheduleException(AirflowException):
"""
+class InvalidBackfillDirection(AirflowException):
+ """
+ Raised when backfill is attempted in reverse with tasks that depend on past runs.
Review Comment:
Done
--
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] Handle backfill for DAGs with when depends_on_past is True [airflow]
Posted by "vatsrahul1001 (via GitHub)" <gi...@apache.org>.
vatsrahul1001 commented on code in PR #45731:
URL: https://github.com/apache/airflow/pull/45731#discussion_r1922644168
##########
airflow/models/backfill.py:
##########
@@ -75,6 +75,22 @@ class DagNoScheduleException(AirflowException):
"""
+class InvalidBackfillDirection(AirflowException):
+ """
+ Raised when backfill is attempted in reverse with tasks that depend on past runs.
+
+ :meta private:
+ """
+
+
+class InvalidReprocessBehavior(AirflowException):
+ """
+ Raised when reprocess behavior is not set for tasks with depends_on_past=True.
Review Comment:
Resolved 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] Handle backfill for DAGs with when depends_on_past is True [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #45731:
URL: https://github.com/apache/airflow/pull/45731#discussion_r1924045795
##########
airflow/api_fastapi/core_api/routes/public/backfills.py:
##########
@@ -216,6 +218,17 @@ def create_backfill(
status_code=status.HTTP_409_CONFLICT,
detail=f"{backfill_request.dag_id} has no schedule",
)
+ except InvalidReprocessBehavior:
+ raise HTTPException(
+ status_code=status.HTTP_409_CONFLICT,
+ detail=f"{backfill_request.dag_id} has tasks for which depends_on_past=True. "
+ "You must set reprocess behavior to reprocess completed or reprocess failed.",
+ )
+ except InvalidBackfillDirection:
+ raise HTTPException(
+ status_code=status.HTTP_409_CONFLICT,
+ detail="Backfill cannot be run in reverse when the DAG has tasks where depends_on_past=True.",
+ )
Review Comment:
Hi @vatsrahul1001, I just tried adding the `__str__` method, and it works fine.
```python
class InvalidReprocessBehavior(AirflowException):
"""
Raised when a backfill cannot be completed because the reprocess behavior is not valid.
:meta private:
"""
def __init__(self, dag_id: str):
self.dag_id = dag_id
self.message = (
f"{dag_id} has tasks for which depends_on_past=True. "
"You must set reprocess behavior to reprocess completed or reprocess failed."
)
super().__init__(self.message)
def __str__(self):
return self.message
# ...
try:
# ...
except InvalidReprocessBehavior as e:
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail=e.message,
)
```
--
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] Handle backfill for DAGs with when depends_on_past is True [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #45731:
URL: https://github.com/apache/airflow/pull/45731#discussion_r1924045795
##########
airflow/api_fastapi/core_api/routes/public/backfills.py:
##########
@@ -216,6 +218,17 @@ def create_backfill(
status_code=status.HTTP_409_CONFLICT,
detail=f"{backfill_request.dag_id} has no schedule",
)
+ except InvalidReprocessBehavior:
+ raise HTTPException(
+ status_code=status.HTTP_409_CONFLICT,
+ detail=f"{backfill_request.dag_id} has tasks for which depends_on_past=True. "
+ "You must set reprocess behavior to reprocess completed or reprocess failed.",
+ )
+ except InvalidBackfillDirection:
+ raise HTTPException(
+ status_code=status.HTTP_409_CONFLICT,
+ detail="Backfill cannot be run in reverse when the DAG has tasks where depends_on_past=True.",
+ )
Review Comment:
Hi @vatsrahul1001, I just tried adding the `__str__` method, and it works fine.
```python
class InvalidReprocessBehavior(AirflowException):
"""
Raised when a backfill cannot be completed because the reprocess behavior is not valid.
:meta private:
"""
def __init__(self, dag_id: str):
self.dag_id = dag_id
self.message = (
f"{dag_id} has tasks for which depends_on_past=True. "
"You must set reprocess behavior to reprocess completed or reprocess failed."
)
super().__init__(self.message)
def __str__(self):
return self.message
# ...
try:
# ...
except InvalidReprocessBehavior as e:
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail=e.message,
)
```
--
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] Handle backfill for DAGs with when depends_on_past is True [airflow]
Posted by "vatsrahul1001 (via GitHub)" <gi...@apache.org>.
vatsrahul1001 commented on code in PR #45731:
URL: https://github.com/apache/airflow/pull/45731#discussion_r1924721037
##########
airflow/models/backfill.py:
##########
@@ -75,6 +75,22 @@ class DagNoScheduleException(AirflowException):
"""
+class InvalidBackfillDirection(AirflowException):
+ """
+ Raised when backfill is attempted in reverse order with tasks that depend on past runs.
+
+ :meta private:
+ """
+
+
+class InvalidReprocessBehavior(AirflowException):
+ """
+ Raised when a backfill cannot be completed because the reprocess behavior is not valid.
+
+ :meta private:
+ """
Review Comment:
Implemented this https://github.com/apache/airflow/pull/45731#discussion_r1924092898
--
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 Handle backfill for DAGs with when depends_on_past is True [airflow]
Posted by "pierrejeambrun (via GitHub)" <gi...@apache.org>.
pierrejeambrun merged PR #45731:
URL: https://github.com/apache/airflow/pull/45731
--
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