merge_machine/abstract_project.py
def upload_config_data(self, config_dict, module_name, file_name):
        '''Write dict type data with a retry logic if a file is locked.'''
        NUM_RETRY = 10
        RETRY_INTERVAL = 0.1
        
        if config_dict is None:
            return

        # Create directories
        dir_path = self.path_to(module_name)
        if (not os.path.isdir(dir_path)) and module_name:
            os.makedirs(dir_path)   

        file_path = self.path_to(module_name, file_name)

        for _ in range(NUM_RETRY):
            try:
                # Lock File before writing
                with open(file_path, 'a') as f:
                    fcntl.flock(f, fcntl.LOCK_EX | fcntl.LOCK_NB)
                
                # Write file
                with open(file_path, 'w') as w:
                    json.dump(config_dict, w, cls=MyEncoder)
                
                    # Unlock file
                    fcntl.flock(w, fcntl.LOCK_UN)
                break
                    
            except BlockingIOError:
                time.sleep(RETRY_INTERVAL)
        else:
            raise BlockingIOError('{0} is un-writable because '.format(file_path) \
                                + 'it was locked for by another process')
Write dict type data with a retry logic if a file is locked.
Write dict type data with a retry logic if a file is locked.
Write dict type data with a retry logic if a file is locked .
def upload_config_data(self, config_dict, module_name, file_name):
        NUM_RETRY = 10
        RETRY_INTERVAL = 0.1
        if config_dict is None:
            return
        dir_path = self.path_to(module_name)
        if (not os.path.isdir(dir_path)) and module_name:
            os.makedirs(dir_path)   
        file_path = self.path_to(module_name, file_name)
        for _ in range(NUM_RETRY):
            try:
                with open(file_path, 'a') as f:
                    fcntl.flock(f, fcntl.LOCK_EX | fcntl.LOCK_NB)
                with open(file_path, 'w') as w:
                    json.dump(config_dict, w, cls=MyEncoder)
                    fcntl.flock(w, fcntl.LOCK_UN)
                break
            except BlockingIOError:
                time.sleep(RETRY_INTERVAL)
        else:
            raise BlockingIOError('{0} is un-writable because '.format(file_path) \
                                + 'it was locked for by another process')
def upload_config_data ( self , config_dict , module_name , file_name ) : NUM_RETRY = 10 RETRY_INTERVAL = 0.1 if config_dict is None : return dir_path = self . path_to ( module_name ) if ( not os . path . isdir ( dir_path ) ) and module_name : os . makedirs ( dir_path ) file_path = self . path_to ( module_name , file_name ) for _ in range ( NUM_RETRY ) : try : with open ( file_path , 'a' ) as f : fcntl . flock ( f , fcntl . LOCK_EX | fcntl . LOCK_NB ) with open ( file_path , 'w' ) as w : json . dump ( config_dict , w , cls = MyEncoder ) fcntl . flock ( w , fcntl . LOCK_UN ) break except BlockingIOError : time . sleep ( RETRY_INTERVAL ) else : raise BlockingIOError ( '{0} is un-writable because ' . format ( file_path ) + 'it was locked for by another process' )
Write dict type data with a retry logic if a file is locked.
Write dict type data with a retry logic if a file is locked .
'''Write dict type data with a retry logic if a file is locked.''' # Create directories # Lock File before writing # Write file # Unlock file
self
config_dict
module_name
file_name