Source code for lambdas.application_dependencies

import logging
import os

from pymorri.util.serverless_utils import resolve_env


[docs]class ApplicationDependencies:
    product_queue_name = None
    product_availability_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:
           Exception
        """
        try:
            logging.basicConfig(level=logging.INFO)
            logging.info("Application starting, loading configurations.....")
            self.product_queue_name = resolve_env(config["PRODUCT_SQS_NAME"])
            self.product_availability_queue_name = resolve_env(config["AVAILABILITY_SQS_NAME"])
            self.product_capping_queue_name = resolve_env(config["CAPPING_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_capping_path = resolve_env(config["S3_PRODUCT_CAPPING_PATH"])
            self.ttl = config["DYNAMO_TTL"]
            self.file_types = config["FILE_TYPES"]
            self.sqs_msg_limit = config["SQS_MSG_LIMIT"]
            self.database = config["DATABASE"]
            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_masking_dynamodb_table(self):
        return self.product_masking_dynamodb_table

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

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

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

[docs]    def get_sqs_msg_limit(self):
        return self.sqs_msg_limit

[docs]    def get_product_availability_queue_name(self):
        return self.product_availability_queue_name

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

[docs]    def get_product_capping_queue_name(self):
        return self.product_capping_queue_name

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