def traverse()

in awscli/customizations/cloudtrail/validation.py [0:0]

• 48 lines of code
• 10 McCabe index (conditional complexity)

    def traverse(self, start_date, end_date=None):
        """Creates and returns a generator that yields validated digest data.

        Each yielded digest dictionary contains information about the digest
        and the log file associated with the digest. Digest files are validated
        before they are yielded. Whether or not the digest is successfully
        validated is stated in the "isValid" key value pair of the yielded
        dictionary.

        :type start_date: datetime
        :param start_date: Date to start validating from (inclusive).
        :type start_date: datetime
        :param end_date: Date to stop validating at (inclusive).
        """
        if end_date is None:
            end_date = datetime.utcnow()
        end_date = normalize_date(end_date)
        start_date = normalize_date(start_date)
        bucket = self.starting_bucket
        prefix = self.starting_prefix
        digests = self._load_digests(bucket, prefix, start_date, end_date)
        public_keys = self._load_public_keys(start_date, end_date)
        key, end_date = self._get_last_digest(digests)
        last_start_date = end_date
        while key and start_date <= last_start_date:
            try:
                digest, end_date = self._load_and_validate_digest(
                    public_keys, bucket, key)
                last_start_date = normalize_date(
                    parse_date(digest['digestStartTime']))
                previous_bucket = digest.get('previousDigestS3Bucket', None)
                yield digest
                if previous_bucket is None:
                    # The chain is broken, so find next in digest store.
                    key, end_date = self._find_next_digest(
                        digests=digests, bucket=bucket, last_key=key,
                        last_start_date=last_start_date, cb=self._on_gap,
                        is_cb_conditional=True)
                else:
                    key = digest['previousDigestS3Object']
                    if previous_bucket != bucket:
                        bucket = previous_bucket
                        # The bucket changed so reload the digest list.
                        digests = self._load_digests(
                            bucket, prefix, start_date, end_date)
            except ClientError as e:
                if e.response['Error']['Code'] != 'NoSuchKey':
                    raise e
                key, end_date = self._find_next_digest(
                    digests=digests, bucket=bucket, last_key=key,
                    last_start_date=last_start_date, cb=self._on_missing,
                    message=str(e))
            except DigestError as e:
                key, end_date = self._find_next_digest(
                    digests=digests, bucket=bucket, last_key=key,
                    last_start_date=last_start_date, cb=self._on_invalid,
                    message=str(e))
            except Exception as e:
                # Any other unexpected errors.
                key, end_date = self._find_next_digest(
                    digests=digests, bucket=bucket, last_key=key,
                    last_start_date=last_start_date, cb=self._on_invalid,
                    message='Digest file\ts3://%s/%s\tINVALID: %s'
                            % (bucket, key, str(e)))