def wait_until_model_status()

in src/core/fraud_detector_utils.py [0:0]

• 19 lines of code
• 7 McCabe index (conditional complexity)

    def wait_until_model_status(self, model_name, model_version, model_type, fail_states, success_states):
        """
        Polls for a model status until either it reaches the failure or success states
        :param model_name: Model name
        :param model_version: Model version
        :param model_type: Model type
        :param fail_states: a single state or a list of states indicating failure.
        :param success_states: a single state or a list of states indicating success.
        :return:
        """
        stime = time.time()
        job_inprogress = True

        # If single state string is passed , convert to list
        if isinstance(fail_states, str): fail_states = [fail_states]
        if isinstance(success_states, str): success_states = [success_states]

        end_states = fail_states + success_states

        while job_inprogress:
            response = self.fraud_detector_client.get_model_version(modelId=model_name, modelType=model_type,
                                                                    modelVersionNumber=str(model_version))

            reponse_status = response['status']
            if reponse_status not in end_states:
                self._logger.info(
                    "Current inprogress, state {}: {:.2f} minutes".format(reponse_status, (time.time() - stime) / 60))
                time.sleep(60)  # -- sleep for 60 seconds
            # Error
            elif reponse_status in fail_states:
                raise Exception("Failed to complete successfully {}".format(response))
            # Complete
            else:
                self._logger.info("Model status : " + reponse_status)
                job_inprogress = False