Source code for ecstask.productavailability.utilities.application_dependencies

import logging
import os

from pymorri.util import SSM
from pymorri.util.aws_utils import get_account_id
from pymorri.util.serverless_utils import resolve_env


[docs]class ApplicationDependencies:
    product_queue_name = None
    picks_table = None
    ttl = None

[docs]    def __init__(self, config):
        """
        This class will load all the dependencies needed for the application to run.
        This can cause exceptions if the configuration is not set up properly.

        Args:
            config (Dictionary): Contains key value pairs for configuration of the google tables
        """
        self.__load_dependencies(config)

    def __load_dependencies(self, config):
        """
        Private Method to load all the dependencies of the application, sort of a dependency injection simulation
        Raises:
            GoogleApplicationCredentialsNotValid
            FlaskConfigurationKeyError
            GoogleApplicationCredentialsNotFound
        """
        try:
            logging.basicConfig(level=logging.INFO)
            logging.info("Application starting, loading configurations.....")
            self.product_queue_name = resolve_env(config["PRODUCT_SQS_NAME"])
            self.checkout_queue_name = resolve_env(config["CHECKOUT_SQS_NAME"])
            self.priority_two_queue_name = resolve_env(config["PRIORITY_TWO_SQS_NAME"])
            self.priority_one_queue_name = resolve_env(config["PRIORITY_ONE_SQS_NAME"])
            self.product_dynamodb_table = resolve_env(config["PRODUCT_DYNAMODB_TABLE"])
            self.product_cap_dynamodb_table = resolve_env(config["PRODUCT_CAP_DYNAMODB_TABLE"])
            self.product_masking_dynamodb_table = resolve_env(config["PRODUCT_MASKING_DYNAMODB_TABLE"])
            self.bulk_upload_dynamodb_table = resolve_env(config["BULK_UPLOAD_DYNAMODB_TABLE"])
            self.audit_dynamodb_table = resolve_env(config["AUDIT_DYNAMODB_TABLE"])
            self.user_role_dynamodb_table = resolve_env(config["USER_ROLES_DYNAMODB_TABLE"])
            self.orders_dynamodb_table = resolve_env(config["ORDERS_DYNAMODB_TABLE"])
            self.s3_bucket_name = resolve_env(config["S3_BUCKET_NAME"])
            self.s3_product_import_path = resolve_env(config["S3_PRODUCT_IMPORT_PATH"])
            self.s3_product_masking_path = resolve_env(config["S3_PRODUCT_AVAILABILITY_PATH"])
            self.s3_product_capping_path = resolve_env(config["S3_PRODUCT_CAPPING_PATH"])
            self.ttl = config["DYNAMO_TTL"]
            self.file_types = config["FILE_TYPES"]
            self.default_fetch_limit = config["DEFAULT_FETCH_LIMIT"]
            self.database = config["DATABASE"]
            self.product_service_secret = SSM('storepick_product_service_secret').get_ssm_credentials()
            self.kafka_broker = config["KAFKA_BROKER"]
            local_credential = os.environ.get("local_cred")
            if local_credential:
                self.kafka_broker = os.environ.get("kafka_broker")
            self.kafka_topic_masking = resolve_env(config['KAFKA_TOPIC_MASKING'])
            self.kafka_topic_capping = resolve_env(config['KAFKA_TOPIC_CAPPING'])

        except Exception as ex:
            logging.error(f"Exception while loading configs {ex}")
            raise ex

[docs]    def get_bootstrap_server(self):
        return self.kafka_broker

[docs]    def get_kafka_masking_topic(self):
        return self.kafka_topic_masking

[docs]    def get_kafka_capping_topic(self):
        return self.kafka_topic_capping

[docs]    def get_product_dynamodb_table(self):
        return self.product_dynamodb_table

[docs]    def get_bulk_upload_dynamodb_table(self):
        return self.bulk_upload_dynamodb_table

    def get_product_masking_dynamodb_table(self):
        return self.product_masking_dynamodb_table

[docs]    def get_ttl(self):
        return self.ttl

[docs]    def get_product_queue_name(self):
        return self.product_queue_name

[docs]    def get_checkout_queue_name(self):
        return self.checkout_queue_name

[docs]    def get_priority_two_queue_name(self):
        return self.priority_two_queue_name

[docs]    def get_priority_one_queue_name(self):
        return self.priority_one_queue_name

[docs]    def get_s3_bucket_name(self):
        return self.s3_bucket_name

[docs]    def get_s3_product_import_path(self):
        return self.s3_product_import_path

[docs]    def get_s3_product_masking_path(self):
        return self.s3_product_masking_path

[docs]    def get_s3_product_capping_path(self):
        return self.s3_product_capping_path

[docs]    def get_file_types(self):
        return self.file_types

[docs]    def get_default_fetch_limit(self):
        return self.default_fetch_limit

[docs]    def get_product_masking_dynamodb_table(self):
        return self.product_masking_dynamodb_table

[docs]    def get_product_cap_dynamodb_table(self):
        return self.product_cap_dynamodb_table

[docs]    def get_audit_dynamodb_table(self):
        return self.audit_dynamodb_table

[docs]    def get_product_service_secret(self):
        return self.product_service_secret