def __process_job__()

in src/edge/app/ota.py [0:0]

• 40 lines of code
• 6 McCabe index (conditional complexity)

    def __process_job__(self, job_id, msg):
        '''
            This method is responsible for:
                1. validate the new model version
                2. download the model package
                3. unpack it to a local dir
                4. notify the main application
        '''
        self.processing_lock.acquire()
        if job_id in self.processed_jobs:
            self.processing_lock.release()
            return
        self.processed_jobs.append(job_id)
        try:
            if msg.get('type') == 'new_model':
                model_version = msg['model_version']
                model_name = msg['model_name']

                # Check if the application supports the model with the name incoming
                if model_name not in self.models_supported:
                    msg = 'New model %s from incoming deployment is not in list of supported models. Skipping deployment.' % model_name
                    logging.info(msg)
                    self.__update_job_status__(job_id, 'FAILED', msg)
                    self.processing_lock.release()
                    return

                logging.info("Downloading new model package")
                s3_client = app.get_client('s3', self.iot_params)

                package = io.BytesIO(s3_client.get_object(
                    Bucket=msg['model_package_bucket'],
                    Key=msg['model_package_key'])['Body'].read()
                )
                logging.info("Unpacking model package")
                with tarfile.open(fileobj=package) as p:
                    p.extractall(os.path.join(self.model_path, msg['model_name'], msg['model_version']))

                self.__update_job_status__(job_id, 'SUCCEEDED', 'Model deployed')
                self.update_callback(model_name, model_version)
            else:
                logging.info("Model '%s' version '%f' is the current one or it is obsolete" % (self.model_metadata['model_name'], self.model_metadata['model_version']))
        except Exception as e:
            self.__update_job_status__(job_id, 'FAILED', str(e))
            logging.error(e)

        self.processing_lock.release()