def list_prefix()

in smdebug/core/access_layer/s3handler.py [0:0]

• 53 lines of code
• 18 McCabe index (conditional complexity)

    def list_prefix(lr):
        count = 0
        while count < S3Handler.NUM_RETRIES:
            try:
                paginator = S3Handler.client().get_paginator("list_objects_v2")
                page_iterator = paginator.paginate(
                    Bucket=lr.bucket,
                    Prefix=lr.prefix,
                    Delimiter=lr.delimiter,
                    StartAfter=lr.start_after,
                    PaginationConfig={"PageSize": 1000},
                )
                keys = []
                for page in page_iterator:
                    if lr.delimiter:
                        if "CommonPrefixes" in page.keys():
                            for pre in page["CommonPrefixes"]:
                                if "Prefix" in pre.keys():
                                    keys += [pre["Prefix"]]
                        if "Contents" in page.keys():
                            for obj in page["Contents"]:
                                if "Key" in obj.keys():
                                    keys += [obj["Key"]]
                    else:
                        if "Contents" in page.keys():
                            for obj in page["Contents"]:
                                if "Key" in obj.keys():
                                    keys += [obj["Key"]]
                return keys
            except (
                NoCredentialsError,  # No credentials could be found
                PartialCredentialsError,
                # Only partial credentials were found.
                CredentialRetrievalError,
                # Error attempting to retrieve credentials from a remote source.
                UnknownSignatureVersionError,
                # Requested Signature Version is not known.
                ServiceNotInRegionError,
                # The service is not available in requested region.
                NoRegionError,  # No region was specified.
                ClientError,
                # Covers cases when the client has insufficient permissions
            ) as botocore_error:
                if isinstance(botocore_error, ClientError):
                    if botocore_error.response["Error"]["Code"] != "AccessDenied":
                        logger.warning(
                            f"Unable to list files "
                            f"from  {lr.bucket}/{str(lr.prefix)}: {str(botocore_error)}"
                        )
                    else:
                        raise botocore_error
                else:
                    raise botocore_error
            except Exception as e:
                logger.warning(str(e))
            count += 1

        # if success, we wouldn't come here
        logger.warning(f"Unable to list files for {lr.bucket} with prefix {lr.prefix}")
        return []