Coverage for /pythoncovmergedfiles/medio/medio/src/airflow/build/lib/airflow/ti_deps/deps/not_in_retry_period_dep.py: 52%
21 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.
18from __future__ import annotations
20from airflow.ti_deps.deps.base_ti_dep import BaseTIDep
21from airflow.utils import timezone
22from airflow.utils.session import provide_session
23from airflow.utils.state import State
26class NotInRetryPeriodDep(BaseTIDep):
27 """Determines whether a task is not in retry period."""
29 NAME = "Not In Retry Period"
30 IGNORABLE = True
31 IS_TASK_DEP = True
33 @provide_session
34 def _get_dep_statuses(self, ti, session, dep_context):
35 if dep_context.ignore_in_retry_period:
36 yield self._passing_status(
37 reason="The context specified that being in a retry period was permitted."
38 )
39 return
41 if ti.state != State.UP_FOR_RETRY:
42 yield self._passing_status(reason="The task instance was not marked for retrying.")
43 return
45 # Calculate the date first so that it is always smaller than the timestamp used by
46 # ready_for_retry
47 cur_date = timezone.utcnow()
48 next_task_retry_date = ti.next_retry_datetime()
49 if ti.is_premature:
50 yield self._failing_status(
51 reason=(
52 f"Task is not ready for retry yet but will be retried automatically. "
53 f"Current date is {cur_date.isoformat()} and task will be retried "
54 f"at {next_task_retry_date.isoformat()}."
55 )
56 )