def apply_restore_config()

in 5-app-infra/3-artifact-publish/docker/cdmc/tag_engine_api/DataCatalogController.py [0:0]

• 58 lines of code
• 14 McCabe index (conditional complexity)

    def apply_restore_config(self, job_uuid, config_uuid, tag_extract, tag_history, overwrite=False):
             
        op_status = constants.SUCCESS
        
        for json_obj in tag_extract:
            #print('json_obj: ', json_obj)
        
            entry_group = json_obj['entryGroupId']
            entry_id = json_obj['id']
            location_id = json_obj['locationId']
            project_id = json_obj['projectId']
    
            #print('entry_group: ', entry_group)
            #print('entry_id: ', entry_id)
        
            entry_name = 'projects/' + project_id + '/locations/' + location_id + '/entryGroups/' + entry_group + '/entries/' + entry_id
            print('entry_name: ', entry_name)
    
            try:
                entry = self.client.get_entry(name=entry_name)
                
            except Exception as e:
                msg = "Error couldn't find the entry: {}".format(entry_name)
                log_error(msg, e, job_uuid)
                op_status = constants.ERROR
                return op_status
            
            if 'columns' in json_obj:
                # column-level tag
                json_columns = json_obj['columns']
                #print('json_columns: ', json_columns)
                
                for column_obj in json_columns:
                
                    column_name = column_obj['name'].split(':')[1]
                    column_tags = column_obj['tags']
                    fields = column_tags[0]['fields']
                                     
                    try:    
                        tag_exists, tag_id = self.check_if_tag_exists(parent=entry.name, column=column_name)
        
                    except Exception as e:
                        msg = 'Error during check_if_tag_exists:{}'.format(entry.name)
                        log_error(msg, e, job_uuid)
                        op_status = constants.ERROR
                        return op_status

                    if tag_exists and overwrite == False:
                        msg = 'Info: Tag already exists and overwrite flag is False'
                        info = {'job_uuid': job_uuid, 'msg': msg}
                        print(json.dumps(info))
                        
                        op_status = constants.SUCCESS
                        return op_status
            
                    # create or update column-level tag
                    uri = entry.linked_resource.replace('//bigquery.googleapis.com/projects/', '') + '/column/' + column_name
                    op_status = self.create_update_delete_tag(fields, tag_exists, tag_id, job_uuid, config_uuid, 'RESTORE_TAG', tag_history, \
                                                                     entry, uri, column_name)
            
            if 'tags' in json_obj:
                # table-level tag
                json_tags = json_obj['tags'] 
                fields = json_tags[0]['fields']
                #print('fields: ', fields)  
                
                try:    
                    tag_exists, tag_id = self.check_if_tag_exists(parent=entry.name, column='')
    
                except Exception as e:
                    msg = 'Error during check_if_tag_exists:{}'.format(entry.name)
                    log_error(msg, e, job_uuid)
                    op_status = constants.ERROR
                    return op_status

                if tag_exists and overwrite == False:
                    msg = 'Info: Tag already exists and overwrite flag is False'
                    info = {'job_uuid': job_uuid, 'msg': msg}
                    print(json.dumps(info))
                    
                    op_status = constants.SUCCESS
                    return op_status
                
                # create or update table-level tag
                uri = entry.linked_resource.replace('//bigquery.googleapis.com/projects/', '')
                op_status = self.create_update_delete_tag(fields, tag_exists, tag_id, job_uuid, config_uuid, 'RESTORE_TAG', tag_history, \
                                                          entry, uri)                     
                    
        return op_status