fix(s3): improve error handling (#2337)

This commit is contained in:
Sergio Garcia
2023-05-10 16:43:06 +02:00
committed by GitHub
parent 9ceaefb663
commit 9943643958

View File

@@ -182,12 +182,10 @@ class S3:
logger.warning( logger.warning(
f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
) )
else: elif (
logger.error( error.response["Error"]["Code"]
f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" == "NoSuchPublicAccessBlockConfiguration"
) ):
except Exception as error:
if "NoSuchPublicAccessBlockConfiguration" in str(error):
# Set all block as False # Set all block as False
bucket.public_access_block = PublicAccessBlock( bucket.public_access_block = PublicAccessBlock(
block_public_acls=False, block_public_acls=False,
@@ -196,14 +194,18 @@ class S3:
restrict_public_buckets=False, restrict_public_buckets=False,
) )
else: else:
if regional_client: logger.error(
logger.error( f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" )
) except Exception as error:
else: if regional_client:
logger.error( logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
) )
else:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
def __get_bucket_acl__(self, bucket): def __get_bucket_acl__(self, bucket):
logger.info("S3 - Get buckets acl...") logger.info("S3 - Get buckets acl...")
@@ -265,22 +267,21 @@ class S3:
logger.warning( logger.warning(
f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
) )
elif error.response["Error"]["Code"] == "OwnershipControlsNotFoundError":
bucket.ownership = None
else: else:
logger.error( logger.error(
f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
) )
except Exception as error: except Exception as error:
if "OwnershipControlsNotFoundError" in str(error): if regional_client:
bucket.ownership = None logger.error(
f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
else: else:
if regional_client: logger.error(
logger.error( f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" )
)
else:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
def __get_object_lock_configuration__(self, bucket): def __get_object_lock_configuration__(self, bucket):
logger.info("S3 - Get buckets ownership controls...") logger.info("S3 - Get buckets ownership controls...")