Coverage for /pythoncovmergedfiles/medio/medio/src/airflow/build/lib/airflow/ti_deps/deps/dagrun_backfill_dep.py: 77%
13 statements
« prev ^ index » next coverage.py v7.2.7, created at 2023-06-07 06:35 +0000
« prev ^ index » next coverage.py v7.2.7, created at 2023-06-07 06:35 +0000
1#
2# Licensed to the Apache Software Foundation (ASF) under one
3# or more contributor license agreements. See the NOTICE file
4# distributed with this work for additional information
5# regarding copyright ownership. The ASF licenses this file
6# to you under the Apache License, Version 2.0 (the
7# "License"); you may not use this file except in compliance
8# with the License. You may obtain a copy of the License at
9#
10# http://www.apache.org/licenses/LICENSE-2.0
11#
12# Unless required by applicable law or agreed to in writing,
13# software distributed under the License is distributed on an
14# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15# KIND, either express or implied. See the License for the
16# specific language governing permissions and limitations
17# under the License.
18"""This module defines dep for making sure DagRun not a backfill."""
19from __future__ import annotations
21from airflow.ti_deps.deps.base_ti_dep import BaseTIDep
22from airflow.utils.session import provide_session
23from airflow.utils.types import DagRunType
26class DagRunNotBackfillDep(BaseTIDep):
27 """Dep for valid DagRun run_id to schedule from scheduler."""
29 NAME = "DagRun is not backfill job"
30 IGNORABLE = True
32 @provide_session
33 def _get_dep_statuses(self, ti, session, dep_context=None):
34 """
35 Determines if the DagRun is valid for scheduling from scheduler.
37 :param ti: the task instance to get the dependency status for
38 :param session: database session
39 :param dep_context: the context for which this dependency should be evaluated for
40 :return: True if DagRun is valid for scheduling from scheduler.
41 """
42 dagrun = ti.get_dagrun(session)
44 if dagrun.run_type == DagRunType.BACKFILL_JOB:
45 yield self._failing_status(
46 reason=f"Task's DagRun run_type is {dagrun.run_type} and cannot be run by the scheduler"
47 )