airflow/contrib/sensors/gcs_sensor.py
def is_bucket_updated(self, current_num_objects):
        """
        Checks whether new objects have been uploaded and the inactivity_period
        has passed and updates the state of the sensor accordingly.

        :param current_num_objects: number of objects in bucket during last poke.
        :type current_num_objects: int
        """

        if current_num_objects > self.previous_num_objects:
            # When new objects arrived, reset the inactivity_seconds
            # previous_num_objects for the next poke.
            self.log.info(
                '''
                New objects found at {} resetting last_activity_time.
                '''.format(os.path.join(self.bucket, self.prefix)))
            self.last_activity_time = get_time()
            self.inactivity_seconds = 0
            self.previous_num_objects = current_num_objects
        elif current_num_objects < self.previous_num_objects:
            # During the last poke interval objects were deleted.
            if self.allow_delete:
                self.previous_num_objects = current_num_objects
                self.last_activity_time = get_time()
                self.log.warning(
                    '''
                    Objects were deleted during the last
                    poke interval. Updating the file counter and
                    resetting last_activity_time.
                    '''
                )
            else:
                raise RuntimeError(
                    '''
                    Illegal behavior: objects were deleted in {} between pokes.
                    '''.format(os.path.join(self.bucket, self.prefix))
                )
        else:
            if self.last_activity_time:
                self.inactivity_seconds = (
                    get_time() - self.last_activity_time).total_seconds()
            else:
                # Handles the first poke where last inactivity time is None.
                self.last_activity_time = get_time()
                self.inactivity_seconds = 0

            if self.inactivity_seconds >= self.inactivity_period:
                if current_num_objects >= self.min_objects:
                    self.log.info(
                        '''
                        SUCCESS:
                        Sensor found {} objects at {}.
                        Waited at least {} seconds, with no new objects dropped.
                        '''.format(
                            current_num_objects,
                            os.path.join(self.bucket, self.prefix),
                            self.inactivity_period))
                    return True

                warn_msg = \
                    '''
                    FAILURE:
                    Inactivity Period passed,
                    not enough objects found in {}
                    '''.format(
                        os.path.join(self.bucket, self.prefix))
                self.log.warning(warn_msg)
                return False
            return False
def is_bucket_updated(self, current_num_objects):
        """
        Checks whether new objects have been uploaded and the inactivity_period
        has passed and updates the state of the sensor accordingly.

        :param current_num_objects: number of objects in bucket during last poke.
        :type current_num_objects: int
        """

        if current_num_objects > self.previous_num_objects:
            # When new objects arrived, reset the inactivity_seconds
            # previous_num_objects for the next poke.
            self.log.info(
                '''
                New objects found at {} resetting last_activity_time.
                '''.format(os.path.join(self.bucket, self.prefix)))
            self.last_activity_time = get_time()
            self.inactivity_seconds = 0
            self.previous_num_objects = current_num_objects
        elif current_num_objects < self.previous_num_objects:
            # During the last poke interval objects were deleted.
            if self.allow_delete:
                self.previous_num_objects = current_num_objects
                self.last_activity_time = get_time()
                self.log.warning(
                    '''
                    Objects were deleted during the last
                    poke interval. Updating the file counter and
                    resetting last_activity_time.
                    '''
                )
            else:
                raise RuntimeError(
                    '''
                    Illegal behavior: objects were deleted in {} between pokes.
                    '''.format(os.path.join(self.bucket, self.prefix))
                )
        else:
            if self.last_activity_time:
                self.inactivity_seconds = (
                    get_time() - self.last_activity_time).total_seconds()
            else:
                # Handles the first poke where last inactivity time is None.
                self.last_activity_time = get_time()
                self.inactivity_seconds = 0

            if self.inactivity_seconds >= self.inactivity_period:
                if current_num_objects >= self.min_objects:
                    self.log.info(
                        '''
                        SUCCESS:
                        Sensor found {} objects at {}.
                        Waited at least {} seconds, with no new objects dropped.
                        '''.format(
                            current_num_objects,
                            os.path.join(self.bucket, self.prefix),
                            self.inactivity_period))
                    return True

                warn_msg = \
                    '''
                    FAILURE:
                    Inactivity Period passed,
                    not enough objects found in {}
                    '''.format(
                        os.path.join(self.bucket, self.prefix))
                self.log.warning(warn_msg)
                return False
            return False
Checks whether new objects have been uploaded and the inactivity_period
        has passed and updates the state of the sensor accordingly.

        :param current_num_objects: number of objects in bucket during last poke.
        :type current_num_objects: int