def _register_actions()

in smdebug/rules/action/action.py [0:0]

• 48 lines of code
• 13 McCabe index (conditional complexity)

    def _register_actions(self, actions_str="", rule_name=""):

        actions_str = actions_str.lower()
        self._logger.info(f"Action string: {actions_str} and rule_name:{rule_name}")
        action_json = json.loads(actions_str)
        actions_list = []
        if isinstance(action_json, dict):
            actions_list.append(action_json)
        elif isinstance(action_json, list):
            actions_list = action_json
        else:
            self._logger.info(
                f"Action string: {actions_str}, expected either a list of dict or dict. Skipping action registering"
            )
            return

        # action : {name:'StopTraining', 'training_job_prefix':''}
        # {name:'sms or email', 'endpoint':''}
        for action_dict in actions_list:
            if not isinstance(action_dict, dict):
                self._logger.info(
                    f"expected dictionary for action, got {action_dict} . Skipping this action."
                )
                continue
            if "name" in action_dict:
                if action_dict["name"] == "stoptraining":
                    training_job_prefix = (
                        action_dict["training_job_prefix"]
                        if "training_job_prefix" in action_dict
                        else None
                    )
                    if training_job_prefix is None:
                        self._logger.info(
                            f"Action :{action_dict['name']} requires 'training_job_prefix'  key to be specified. "
                            f"Action_dict is: {action_dict}"
                        )
                        continue

                    action = StopTrainingAction(rule_name, training_job_prefix)
                    self._actions.append(action)
                elif action_dict["name"] == "sms" or action_dict["name"] == "email":
                    endpoint = action_dict["endpoint"] if "endpoint" in action_dict else None
                    if endpoint is None:
                        self._logger.info(
                            f"Action :{action_dict['name']} requires endpoint key parameter. "
                        )
                        continue

                    action = MessageAction(rule_name, action_dict["name"], endpoint)
                    self._actions.append(action)

                else:
                    self._logger.info(
                        f"Action :{action_dict['name']} not supported. Allowed action names are: {ALLOWED_ACTIONS}"
                    )