notifications-api/app/cloudfoundry_config.py

import json
from os import getenv


class CloudfoundryConfig:
    def __init__(self):
        self.parsed_services = json.loads(getenv('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 database_url(self):
        return getenv('DATABASE_URL', '').replace('postgres://', 'postgresql://')

    @property
    def redis_url(self):
        try:
            return self.parsed_services['aws-elasticache-redis'][0]['credentials']['uri'].replace(
                'redis://',
                'rediss://'
            )
        except KeyError:
            return getenv('REDIS_URL')

    def s3_credentials(self, service_name):
        return self.s3_buckets.get(service_name) or self._empty_bucket_credentials

    @property
    def ses_email_domain(self):
        try:
            return self._ses_credentials('domain_arn').split('/')[-1]
        except KeyError:
            return getenv('NOTIFY_EMAIL_DOMAIN', 'notify.sandbox.10x.gsa.gov')

    @property
    def ses_region(self):
        try:
            return self._ses_credentials('region')
        except KeyError:
            return getenv('AWS_REGION')

    @property
    def ses_access_key(self):
        try:
            return self._ses_credentials('smtp_user')
        except KeyError:
            return getenv('AWS_ACCESS_KEY_ID')

    @property
    def ses_secret_key(self):
        try:
            return self._ses_credentials('secret_access_key')
        except KeyError:
            return getenv('AWS_SECRET_ACCESS_KEY')

    @property
    def sns_access_key(self):
        try:
            return self._sns_credentials('aws_access_key_id')
        except KeyError:
            return getenv('AWS_ACCESS_KEY_ID')

    @property
    def sns_secret_key(self):
        try:
            return self._sns_credentials('aws_secret_access_key')
        except KeyError:
            return getenv('AWS_SECRET_ACCESS_KEY')

    @property
    def sns_region(self):
        try:
            return self._sns_credentials('region')
        except KeyError:
            return getenv('AWS_REGION')

    @property
    def sns_topic_arns(self):
        try:
            return [
                self._ses_credentials('bounce_topic_arn'),
                self._ses_credentials('complaint_topic_arn'),
                self._ses_credentials('delivery_topic_arn')
            ]
        except KeyError:
            return []

    def _ses_credentials(self, key):
        return self.parsed_services['datagov-smtp'][0]['credentials'][key]

    def _sns_credentials(self, key):
        return self.parsed_services['ttsnotify-sms'][0]['credentials'][key]


cloud_config = CloudfoundryConfig()