Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
data=data,
json=json,
headers=headers,
files=files,
timeout=timeout,
)
except requests.exceptions.ConnectionError as e:
logger.debug("Exception: {}".format(e, exc_info=True))
sys.exit("Connection error. Please retry or check your internet connection")
except requests.exceptions.Timeout as e:
logger.debug("Exception: {}".format(e, exc_info=True))
sys.exit(
"Connection timeout. Please retry or check your internet connection"
)
logger.debug(
"Response Content: {}, Headers: {}".format(
response.content, response.headers
)
)
self.check_response_status(response)
return response
def put(self, path, content):
path = os.path.join(self.bucket, path)
try:
folder = '/'.join(path.split('/')[:-1])
if not os.path.isdir(folder):
os.makedirs(folder)
with open(path, 'wb') as f:
f.write(content)
except IOError as err:
logger.debug(err)
raise FileSystemException
def check_response_status(self, response):
"""
Check response status and throw corresponding exception on failure
"""
code = response.status_code
if code < 200 or code >= 300:
try:
message = response.json()["error"]
except Exception:
message = " "
logger.debug(
'Error received: status code: {}, message: "{}"'.format(code, message)
)
if code == 400:
raise BadRequestException(response)
elif response.status_code == 401:
raise AuthenticationException()
elif response.status_code == 403:
raise AuthorizationException()
elif response.status_code == 404:
raise NotFoundException()
elif response.status_code == 429:
raise OverLimitException(message)
elif response.status_code == 502:
raise BadGatewayException()
elif response.status_code == 504:
raise GatewayTimeoutException(message)
def configure(username, password, bucket):
""" Logs in to Hub"""
logger.info("Please log in using your AWS credentials.")
if not username:
logger.debug("Prompting for Access Key")
username = click.prompt('AWS Access Key ID', type=str, hide_input=False)
access_key = username.strip()
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)