Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def exist_bucket(self, bucket):
try:
response = self.client.list_buckets()
except ClientError as e:
logger.error(e)
return
for bucket in response['Buckets']:
if bucket["Name"] == bucket:
return bucket
return ''
def lookup_hub_bucket(self):
try:
response = self.client.list_buckets()
for bucket in response['Buckets']:
if 'snark-hub-' in bucket["Name"]:
return bucket["Name"]
except ClientError as e:
logger.error(e)
return None
def put(self, path, content, content_type, compress=False, cache_control=False):
try:
attrs = {
'Bucket': self.bucket,
'Body': content,
'Key': path.replace('s3://{}/'.format(self.bucket), ''),
'ContentType': (content_type or 'application/octet-stream'),
}
if compress:
attrs['ContentEncoding'] = 'gzip'
if cache_control:
attrs['CacheControl'] = cache_control
self.client.put_object(**attrs)
except Exception as err:
logger.error(err)
raise
if not password:
logger.debug("Prompting for Secret Key")
password = click.prompt('AWS Secret Access Key', type=str, hide_input=False)
secret_key = password.strip()
if not bucket:
logger.debug("Prompting for bucket name")
bucket = click.prompt('Bucket Name (e.g. company-name)', type=str, hide_input=False)
bucket = bucket.strip()
success, creds = Verify(access_key, secret_key).verify_aws(bucket)
if success:
StoreControlClient().save_config(creds)
logger.info("Login Successful.")
else:
logger.error("Login error, please try again")
try:
attrs = {
'Bucket': self.bucket,
'Body': content,
'Key': path,
'ContentType': ('application/octet-stream'),
}
# if compress:
# attrs['ContentEncoding'] = 'gzip'
# if cache_control:
# attrs['CacheControl'] = cache_control
self.client.put_object(**attrs)
except Exception as err:
logger.error(err)
raise S3Exception(err)
def get(self, path):
try:
resp = self.client.get_object(
Bucket=self.bucket,
Key=path.replace('s3://{}/'.format(self.bucket), ''),
)
return resp['Body'].read()
except botocore.exceptions.ClientError as err:
if err.response['Error']['Code'] == 'NoSuchKey':
return None
else:
logger.error(err)
raise
def get(self, path):
path = path.replace('s3://{}/'.format(self.bucket), '')
try:
resp = self.client.get_object(
Bucket=self.bucket,
Key=path,
)
return resp['Body'].read()
except botocore.exceptions.ClientError as err:
if err.response['Error']['Code'] == 'NoSuchKey':
return None
else:
logger.error(err)
raise S3Exception(err)
:param bucket_name: Bucket to create
:param region: String region to create bucket in, e.g., 'us-west-2'
:return: True if bucket created, else False
"""
# Create bucket
try:
if region is None:
self.client.create_bucket(Bucket=bucket_name)
else:
location = {'LocationConstraint': region}
self.client.create_bucket(Bucket=bucket_name,
CreateBucketConfiguration=location)
except ClientError as e:
logger.error(e)
return False
return True