o-nikolas commented on code in PR #53215:
URL: https://github.com/apache/airflow/pull/53215#discussion_r2208067083
##########
airflow-core/src/airflow/models/deadline.py:
##########
@@ -98,6 +98,61 @@ def _determine_resource() -> tuple[str, str]:
f"{self.deadline_time} or run: {self.callback}({callback_kwargs})"
)
+ @classmethod
+ @provide_session
+ def remove_deadlines(cls, *, session: Session, conditions: dict[Column,
Any]) -> int:
+ """
+ Remove deadlines from the table which match the provided conditions
and return the number removed.
+
+ NOTE: This should only be used to remove deadlines which are
associated with
+ successful dagruns. If the deadline was missed, move it to the
`missed_deadlines`
+ table after executing the callback.
+ TODO: Create the missed_deadlines table (Ramit)
+
+ :param conditions: Dictionary of conditions to evaluate against.
+ :param session: Session to use.
+ """
+ from airflow.models import DagRun # Avoids circular import
+
+ # Assemble the filter conditions.
+ filter_conditions = [column == value for column, value in
conditions.items()]
+ if not filter_conditions:
+ return 0
+
+ try:
+ # Get deadlines which match the provided conditions and their
associated DagRuns.
+ deadline_dagrun_pairs = (
+ session.query(Deadline,
DagRun).join(DagRun).filter(and_(*filter_conditions)).all()
+ )
+ except SQLAlchemyError as e:
+ invalid_column = next(iter(conditions.keys())) # Get the first
key that caused the error.
+ message = f"Invalid column '{invalid_column}' specified in
conditions while resolving deadlines. Rolling back changes."
+ logger.exception(message)
+ session.rollback()
Review Comment:
What's being rolled back here? Wasn't there just a query not a
write/transaction?
##########
airflow-core/src/airflow/models/deadline.py:
##########
@@ -98,6 +98,61 @@ def _determine_resource() -> tuple[str, str]:
f"{self.deadline_time} or run: {self.callback}({callback_kwargs})"
)
+ @classmethod
+ @provide_session
+ def remove_deadlines(cls, *, session: Session, conditions: dict[Column,
Any]) -> int:
+ """
+ Remove deadlines from the table which match the provided conditions
and return the number removed.
+
+ NOTE: This should only be used to remove deadlines which are
associated with
+ successful dagruns. If the deadline was missed, move it to the
`missed_deadlines`
+ table after executing the callback.
Review Comment:
Is the move intended to happen here? I wonder if we should rename this thing
to something like `process_deadlines` since it feels like a bit more than
deletion might happen here.
##########
airflow-core/src/airflow/models/dagrun.py:
##########
@@ -1217,6 +1218,10 @@ def recalculate(self) -> _UnfinishedStates:
msg="success",
)
+ if (deadline := dag.deadline) and isinstance(deadline.reference,
DeadlineReference.TYPES.DAGRUN):
+ # The dagrun has succeeded, so the deadline is no longer
needed.
+ Deadline.remove_deadlines(session=session,
conditions={DagRun.run_id: self.run_id})
Review Comment:
Why are `conditions` being exposed here? Seems just a tiny bit strange.
Shouldn't this just take a param `run_id=self.run_id` and the conditions can be
built within the function? Not sure why that's leaking out to the caller.
##########
airflow-core/tests/unit/models/test_deadline.py:
##########
@@ -103,6 +105,50 @@ def test_add_deadline(self, dagrun, session):
assert result.callback == deadline_orm.callback
assert result.callback_kwargs == deadline_orm.callback_kwargs
+ @pytest.mark.parametrize(
+ "conditions",
+ [
+ pytest.param({}, id="empty_conditions"),
+ pytest.param({Deadline.dagrun_id: INVALID_RUN_ID},
id="no_matches"),
+ pytest.param({Deadline.dagrun_id: RUN_ID}, id="single_condition"),
+ pytest.param({Deadline.dagrun_id: RUN_ID, Deadline.dag_id:
DAG_ID}, id="multiple_conditions"),
+ pytest.param(
+ {Deadline.dagrun_id: RUN_ID, Deadline.dag_id: INVALID_DAG_ID},
id="mixed_conditions"
+ ),
+ ],
+ )
+ @mock.patch("sqlalchemy.orm.Session")
+ def test_resolve_deadlines(self, mock_session, conditions):
Review Comment:
```suggestion
def test_remove_deadlines(self, mock_session, conditions):
```
The code under test seems to be the remove right? Same for below
##########
airflow-core/src/airflow/models/dagrun.py:
##########
@@ -1217,6 +1218,10 @@ def recalculate(self) -> _UnfinishedStates:
msg="success",
)
+ if (deadline := dag.deadline) and isinstance(deadline.reference,
DeadlineReference.TYPES.DAGRUN):
+ # The dagrun has succeeded, so the deadline is no longer
needed.
Review Comment:
```suggestion
# The dagrun has succeeded, so the deadline(s) may no longer
be needed.
```
Or just completely refactor.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
