def config_changed()

in connectors/agent/config.py [0:0]

• 47 lines of code
• 11 McCabe index (conditional complexity)

    def config_changed(self, new_config):
        """See if configuration passed in new_config will update currently stored configuration

        This method takes the new configuration received from the agent and see if there are any changes
        to existing configuration.

        If new_config contains new values for relevant fields, then True is returned, otherwise it returns False.
        """
        # TODO: For now manually check, need to think of a better way?
        # Not super proud of this function, but hey it's tested
        logger.debug("Checking if config changed")
        current_config = self._default_config.copy()
        current_config.update(self.specific_config)

        def _log_level_changed():
            new_config_log_level = nested_get_from_dict(
                new_config, ("service", "log_level")
            )
            current_config_log_level = nested_get_from_dict(
                current_config, ("service", "log_level")
            )

            if new_config_log_level is None:
                return False

            return current_config_log_level != new_config_log_level

        def _elasticsearch_config_changed():
            return current_config.get("elasticsearch") != new_config.get(
                "elasticsearch"
            )

        def _connectors_config_changes():
            current_connectors = current_config.get("connectors", [])
            new_connectors = new_config.get("connectors", [])

            if len(current_connectors) != len(new_connectors):
                return True

            current_connectors_dict = {
                connector["connector_id"]: connector for connector in current_connectors
            }
            new_connectors_dict = {
                connector["connector_id"]: connector for connector in new_connectors
            }

            if set(current_connectors_dict.keys()) != set(new_connectors_dict.keys()):
                return True

            for connector_id in current_connectors_dict:
                current_connector = current_connectors_dict[connector_id]
                new_connector = new_connectors_dict[connector_id]

                if current_connector != new_connector:
                    return True

            return False

        if _log_level_changed():
            logger.debug("log_level changed")
            return True

        if _elasticsearch_config_changed():
            logger.debug("elasticsearch changed")
            return True

        if _connectors_config_changes():
            logger.debug("connectors changed")
            return True

        return False