Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _get_state_file_from_s3(self, state_file_url, profile=None, region=None):
if profile:
session = boto3.session.Session(profile_name=profile, region_name=region)
else:
session = get_boto3_session()
s3 = session.resource('s3')
parts = state_file_url[5:].split('/')
bucket = parts[0]
filename = "/".join(parts[1:])
key = s3.Object(bucket, filename)
try:
state_file = key.get()["Body"].read().decode('utf-8')
except ClientError as ex:
if ex.response['Error']['Code'] == 'NoSuchKey':
raise NoSuchStateFile("Could not find Terraform state file {}".format(self.state_file_url))
else:
raise ex
return json.loads(state_file)
def wrapper(*args, **kwargs):
try:
args[0].obj['CONFIG'] = Config(
filename=args[0].obj['CONFIG_FILE'],
env_file=args[0].obj['ENV_FILE'],
import_env=args[0].obj['IMPORT_ENV'],
tfe_token=args[0].obj['TFE_TOKEN']
)
except NoSuchStateFile as e:
click.echo(str(e))
sys.exit(1)
else:
return func(*args, **kwargs)
return wrapper