def execute(self, context: Dict) -> Any:
    started_at = timezone.utcnow()
    try_number = 1
    log_dag_id = (self.dag.dag_id if self.has_dag() else '')
    if self.reschedule:
        task_reschedules = TaskReschedule.find_for_task_instance(context['ti'])
        if task_reschedules:
            started_at = task_reschedules[0].start_date
            try_number = (len(task_reschedules) + 1)
    while (not self.poke(context)):
        if ((timezone.utcnow() - started_at).total_seconds() > self.timeout):
            if (self.soft_fail and (not context['ti'].is_eligible_to_retry())):
                self._do_skip_downstream_tasks(context)
                raise AirflowSkipException(f'Snap. Time is OUT. DAG id: {log_dag_id}')
            else:
                raise AirflowSensorTimeout(f'Snap. Time is OUT. DAG id: {log_dag_id}')
        if self.reschedule:
            reschedule_date = (timezone.utcnow() + timedelta(seconds=self._get_next_poke_interval(started_at, try_number)))
            raise AirflowRescheduleException(reschedule_date)
        else:
            sleep(self._get_next_poke_interval(started_at, try_number))
            try_number += 1
    self.log.info('Success criteria met. Exiting.')
def execute(self, context: Dict) -> Any:
    started_at = timezone.utcnow()
    try_number = 1
    log_dag_id = (self.dag.dag_id if self.has_dag() else )
    if self.reschedule:
        task_reschedules = TaskReschedule.find_for_task_instance(context['ti'])
        if task_reschedules:
            started_at = task_reschedules[0].start_date
            try_number = (len(task_reschedules) + 1)
    while (not self.poke(context)):
        if ((timezone.utcnow() - started_at).total_seconds() > self.timeout):
            if (self.soft_fail and (not context['ti'].is_eligible_to_retry())):
                self._do_skip_downstream_tasks(context)
                raise AirflowSkipException(f'Snap. Time is OUT. DAG id: {log_dag_id}')
            else:
                raise AirflowSensorTimeout(f'Snap. Time is OUT. DAG id: {log_dag_id}')
        if self.reschedule:
            reschedule_date = (timezone.utcnow() + timedelta(seconds=self._get_next_poke_interval(started_at, try_number)))
            raise AirflowRescheduleException(reschedule_date)
        else:
            sleep(self._get_next_poke_interval(started_at, try_number))
            try_number += 1
    self.log.info('Success criteria met. Exiting.')