Coverage for /pythoncovmergedfiles/medio/medio/src/airflow/airflow/ti_deps/deps/dagrun_backfill_dep.py: 77%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
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."""
20from __future__ import annotations
22from airflow.ti_deps.deps.base_ti_dep import BaseTIDep
23from airflow.utils.session import provide_session
24from airflow.utils.types import DagRunType
27class DagRunNotBackfillDep(BaseTIDep):
28 """Dep for valid DagRun run_id to schedule from scheduler."""
30 NAME = "DagRun is not backfill job"
31 IGNORABLE = True
33 @provide_session
34 def _get_dep_statuses(self, ti, session, dep_context=None):
35 """
36 Determine if the DagRun is valid for scheduling from scheduler.
38 :param ti: the task instance to get the dependency status for
39 :param session: database session
40 :param dep_context: the context for which this dependency should be evaluated for
41 :return: True if DagRun is valid for scheduling from scheduler.
42 """
43 dagrun = ti.get_dagrun(session)
45 if dagrun.run_type == DagRunType.BACKFILL_JOB:
46 yield self._failing_status(
47 reason=f"Task's DagRun run_type is {dagrun.run_type} and cannot be run by the scheduler"
48 )