<s> import os . path <EOL> import subprocess <EOL> from release . storage import AbstractStorageProvider <EOL> class LocalStorageProvider ( AbstractStorageProvider ) : <EOL> name = '<STR_LIT>' <EOL> def __init__ ( self , path ) : <EOL> assert isinstance ( path , str ) <EOL> assert not path . endswith ( '<STR_LIT:/>' ) <EOL> self . __storage_path = path <EOL> def __full_path ( self , path ) : <EOL> return self . __storage_path + '<STR_LIT:/>' + path <EOL> def fetch ( self , path ) : <EOL> with open ( self . __full_path ( path ) , '<STR_LIT:rb>' ) as f : <EOL> return f . read ( ) <EOL> def download_inner ( self , path , local_path ) : <EOL> subprocess . check_call ( [ '<STR_LIT>' , self . __full_path ( path ) , local_path ] ) <EOL> def __copy ( self , full_source_path , full_destination_path ) : <EOL> subprocess . check_call ( [ '<STR_LIT>' , '<STR_LIT>' , os . path . dirname ( full_destination_path ) ] ) <EOL> subprocess . check_call ( [ '<STR_LIT>' , full_source_path , full_destination_path ] ) <EOL> def copy ( self , source_path , destination_path ) : <EOL> self . __copy ( self . __full_path ( source_path ) , self . __full_path ( destination_path ) ) <EOL> def upload ( self , destination_path , blob = None , local_path = None , no_cache = None , content_type = None ) : <EOL> destination_full_path = self . __full_path ( destination_path ) <EOL> subprocess . check_call ( [ '<STR_LIT>' , '<STR_LIT>' , os . path . dirname ( destination_full_path ) ] ) <EOL> assert local_path is None or blob is None <EOL> if local_path : <EOL> self . __copy ( local_path , destination_full_path ) <EOL> else : <EOL> assert isinstance ( blob , bytes ) <EOL> with open ( destination_full_path , '<STR_LIT:wb>' ) as f : <EOL> f . write ( blob ) <EOL> def exists ( self , path ) : <EOL> assert path [ <NUM_LIT:0> ] != '<STR_LIT:/>' <EOL> return os . path . exists ( self . __full_path ( path ) ) <EOL> def remove_recursive ( self , path ) : <EOL> full_path = self . __full_path ( path ) <EOL> assert len ( path ) > <NUM_LIT:5> <EOL> assert len ( full_path ) > <NUM_LIT:5> <EOL> subprocess . check_call ( [ '<STR_LIT>' , '<STR_LIT>' , full_path ] ) <EOL> def list_recursive ( self , path ) : <EOL> final_filenames = set ( ) <EOL> for dirpath , _ , filenames in os . walk ( self . __full_path ( path ) ) : <EOL> assert dirpath . startswith ( self . __storage_path ) <EOL> dirpath_no_prefix = dirpath [ len ( self . __storage_path ) + <NUM_LIT:1> : ] <EOL> for filename in filenames : <EOL> final_filenames . add ( dirpath_no_prefix + '<STR_LIT:/>' + filename ) <EOL> return final_filenames <EOL> @ property <EOL> def url ( self ) : <EOL> return '<STR_LIT>' + self . __storage_path + '<STR_LIT:/>' <EOL> factories = { <EOL> '<STR_LIT:path>' : LocalStorageProvider <EOL> } </s>
