def __init__()

in CommonLayerCode/datalake-library/python/datalake_library/octagon/config.py [0:0]

• 38 lines of code
• 9 McCabe index (conditional complexity)

    def __init__(self, config_file, instance):
        self.logger = logging.getLogger(__name__)
        self.instance = instance
        self.logger.debug(f"Reading configuration from file {config_file}")

        if not os.path.isfile(config_file):
            msg = f"Octagon configuration file is not found {config_file}"
            self.logger.error(msg)
            raise ValueError(msg)

        with open(config_file, "r") as f:
            config_dict = json.load(f)

        self.table_info = {}
        self.metric_info = []

        for config_instance in config_dict["configuration_instances"]:
            if config_instance["instance"] == instance:
                for ti in config_instance["tables"]:
                    if "object" in ti.keys() and "table_name" in ti.keys():
                        table_info = ConfigTableInfo(
                            dynamo_table_name=ti["table_name"],
                            ttl_in_days=ti.get("ttl", 0),
                            read_capacity=ti.get("read_capacity", 0),
                            write_capacity=ti.get("write_capacity", 0),
                        )
                        object_name = ti["object"]

                        self.table_info[object_name] = table_info
                        self.logger.debug(f"Loaded config for table: {table_info}")

                if "metrics" in config_instance.keys():
                    for mi in config_instance["metrics"]:
                        metric_info = MetricInfo(
                            metric=mi.get("metric"),
                            evaluation=mi.get("evaluation"),
                            threshold=mi.get("threshold"),
                            notify=mi.get("notify"),
                            metric_type=mi.get("metric_type"),
                            sns_topic=mi.get("sns_topic", ""),
                        )

                        self.metric_info.append(metric_info)
                        self.logger.debug(f"Loaded config for metric: {metric_info}")

        throw_if_false(len(self.table_info) > 0, "Configuration instance is not found")