yuqian90 commented on a change in pull request #7276: [AIRFLOW-5391] Do not run
skipped tasks when they are cleared
URL: https://github.com/apache/airflow/pull/7276#discussion_r380155542
##########
File path: airflow/models/skipmixin.py
##########
@@ -16,42 +16,38 @@
# specific language governing permissions and limitations
# under the License.
-from typing import Iterable, Set, Union
+from typing import Iterable, Optional, Set, Union
from airflow.models.taskinstance import TaskInstance
from airflow.utils import timezone
from airflow.utils.log.logging_mixin import LoggingMixin
-from airflow.utils.session import provide_session
+from airflow.utils.session import create_session, provide_session
from airflow.utils.state import State
+XCOM_SKIPMIXIN_KEY = "skipmixin_key"
+
class SkipMixin(LoggingMixin):
- @provide_session
- def skip(self, dag_run, execution_date, tasks, session=None):
+ def _set_state_to_skipped(self, dag_run, execution_date, tasks, session):
"""
- Sets tasks instances to skipped from the same dag run.
-
- :param dag_run: the DagRun for which to set the tasks to skipped
- :param execution_date: execution_date
- :param tasks: tasks to skip (not task_ids)
- :param session: db session to use
+ Used internally to set state of task instances to skipped from the
same dag run.
"""
- if not tasks:
- return
-
task_ids = [d.task_id for d in tasks]
now = timezone.utcnow()
if dag_run:
session.query(TaskInstance).filter(
TaskInstance.dag_id == dag_run.dag_id,
TaskInstance.execution_date == dag_run.execution_date,
- TaskInstance.task_id.in_(task_ids)
- ).update({TaskInstance.state: State.SKIPPED,
- TaskInstance.start_date: now,
- TaskInstance.end_date: now},
- synchronize_session=False)
- session.commit()
Review comment:
Yes. Thanks. I refactored the shared code into `_set_state_to_skipped()`
and moved the `commit()` into the caller `skip()` and `skip_all_except()`.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
