Refactor method of retrieving credentials out of VCAP_SERVICES

This commit is contained in:
Ryan Ahearn
2022-10-20 08:03:32 -04:00
parent d87d673b85
commit 921c872bd1
10 changed files with 132 additions and 178 deletions

View File

@@ -2,42 +2,30 @@ import json
import os
def find_by_service_name(services, service_name):
for i in range(len(services)):
if services[i]['name'] == service_name:
return services[i]
return None
class CloudfoundryConfig:
def __init__(self):
self.parsed_services = json.loads(os.environ.get('VCAP_SERVICES') or '{}')
buckets = self.parsed_services.get('s3') or []
self.s3_buckets = {bucket['name']: bucket['credentials'] for bucket in buckets}
self._empty_bucket_credentials = {
'bucket': '',
'access_key_id': '',
'secret_access_key': '',
'region': ''
}
@property
def redis_url(self):
try:
return self.parsed_services['aws-elasticache-redis'][0]['credentials']['uri'].replace(
'redis://',
'rediss://'
)
except KeyError:
return os.environ.get('REDIS_URL')
def s3_credentials(self, service_name):
return self.s3_buckets.get(service_name) or self._empty_bucket_credentials
def extract_cloudfoundry_config():
vcap_services = json.loads(os.environ['VCAP_SERVICES'])
# Redis config
os.environ['REDIS_URL'] = vcap_services['aws-elasticache-redis'][0]['credentials']['uri'].replace('redis', 'rediss')
# CSV Upload Bucket Name
bucket_service = find_by_service_name(
vcap_services['s3'], f"notifications-api-csv-upload-bucket-{os.environ['DEPLOY_ENV']}")
if bucket_service:
os.environ['CSV_UPLOAD_BUCKET_NAME'] = bucket_service['credentials']['bucket']
os.environ['CSV_UPLOAD_ACCESS_KEY'] = bucket_service['credentials']['access_key_id']
os.environ['CSV_UPLOAD_SECRET_KEY'] = bucket_service['credentials']['secret_access_key']
os.environ['CSV_UPLOAD_REGION'] = bucket_service['credentials']['region']
# Contact List Bucket Name
bucket_service = find_by_service_name(
vcap_services['s3'], f"notifications-api-contact-list-bucket-{os.environ['DEPLOY_ENV']}")
if bucket_service:
os.environ['CONTACT_LIST_BUCKET_NAME'] = bucket_service['credentials']['bucket']
os.environ['CONTACT_LIST_ACCESS_KEY'] = bucket_service['credentials']['access_key_id']
os.environ['CONTACT_LIST_SECRET_KEY'] = bucket_service['credentials']['secret_access_key']
os.environ['CONTACT_LIST_REGION'] = bucket_service['credentials']['region']
# Logo Upload Bucket Name
bucket_service = find_by_service_name(
vcap_services['s3'], f"notifications-admin-logo-upload-bucket-{os.environ['DEPLOY_ENV']}")
if bucket_service:
os.environ['LOGO_UPLOAD_BUCKET_NAME'] = bucket_service['credentials']['bucket']
os.environ['LOGO_UPLOAD_ACCESS_KEY'] = bucket_service['credentials']['access_key_id']
os.environ['LOGO_UPLOAD_SECRET_KEY'] = bucket_service['credentials']['secret_access_key']
os.environ['LOGO_UPLOAD_REGION'] = bucket_service['credentials']['region']
cloud_config = CloudfoundryConfig()