<s> """<STR_LIT>""" <EOL> import logging <EOL> import os <EOL> from core . platform import models <EOL> import feconf <EOL> import utils <EOL> ( file_models , ) = models . Registry . import_models ( [ <EOL> models . NAMES . file <EOL> ] ) <EOL> CHANGE_LIST_SAVE = [ { '<STR_LIT>' : '<STR_LIT>' } ] <EOL> class FileMetadata ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , metadata ) : <EOL> self . _size = metadata . size if ( metadata is not None ) else None <EOL> @ property <EOL> def size ( self ) : <EOL> return self . _size <EOL> class FileStreamWithMetadata ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , content , version , metadata ) : <EOL> """<STR_LIT>""" <EOL> self . _content = content <EOL> self . _version = version <EOL> self . _metadata = FileMetadata ( metadata ) <EOL> def read ( self ) : <EOL> """<STR_LIT>""" <EOL> content = self . _content <EOL> self . _content = '<STR_LIT>' <EOL> return content <EOL> @ property <EOL> def metadata ( self ) : <EOL> return self . _metadata <EOL> @ property <EOL> def version ( self ) : <EOL> return self . _version <EOL> class ExplorationFileSystem ( object ) : <EOL> """<STR_LIT>""" <EOL> _DEFAULT_VERSION_NUMBER = <NUM_LIT:1> <EOL> def __init__ ( self , exploration_id ) : <EOL> self . _exploration_id = exploration_id <EOL> @ property <EOL> def exploration_id ( self ) : <EOL> return self . _exploration_id <EOL> def _get_file_metadata ( self , filepath , version ) : <EOL> """<STR_LIT>""" <EOL> if version is None : <EOL> return file_models . FileMetadataModel . get_model ( <EOL> self . _exploration_id , '<STR_LIT>' % filepath ) <EOL> else : <EOL> return file_models . FileMetadataModel . get_version ( <EOL> self . _exploration_id , '<STR_LIT>' % filepath , version ) <EOL> def _get_file_data ( self , filepath , version ) : <EOL> """<STR_LIT>""" <EOL> if version is None : <EOL> return file_models . FileModel . get_model ( <EOL> self . _exploration_id , '<STR_LIT>' % filepath ) <EOL> else : <EOL> return file_models . FileModel . get_version ( <EOL> self . _exploration_id , '<STR_LIT>' % filepath , version ) <EOL> def _save_file ( self , user_id , filepath , raw_bytes ) : <EOL> """<STR_LIT>""" <EOL> if len ( raw_bytes ) > feconf . MAX_FILE_SIZE_BYTES : <EOL> raise Exception ( '<STR_LIT>' ) <EOL> metadata = self . _get_file_metadata ( filepath , None ) <EOL> if not metadata : <EOL> metadata = file_models . FileMetadataModel . create ( <EOL> self . _exploration_id , '<STR_LIT>' % filepath ) <EOL> metadata . size = len ( raw_bytes ) <EOL> data = self . _get_file_data ( filepath , None ) <EOL> if not data : <EOL> data = file_models . FileModel . create ( <EOL> self . _exploration_id , '<STR_LIT>' % filepath ) <EOL> data . content = raw_bytes <EOL> data . commit ( user_id , CHANGE_LIST_SAVE ) <EOL> metadata . commit ( user_id , CHANGE_LIST_SAVE ) <EOL> def get ( self , filepath , version = None , mode = None ) : <EOL> """<STR_LIT>""" <EOL> metadata = self . _get_file_metadata ( filepath , version ) <EOL> if metadata : <EOL> data = self . _get_file_data ( filepath , version ) <EOL> if data : <EOL> if version is None : <EOL> version = data . version <EOL> return FileStreamWithMetadata ( data . content , version , metadata ) <EOL> else : <EOL> logging . error ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' % ( filepath , version ) ) <EOL> return None <EOL> else : <EOL> return None <EOL> def commit ( self , user_id , filepath , raw_bytes ) : <EOL> """<STR_LIT>""" <EOL> self . _save_file ( user_id , filepath , raw_bytes ) <EOL> def delete ( self , user_id , filepath ) : <EOL> """<STR_LIT>""" <EOL> metadata = self . _get_file_metadata ( filepath , None ) <EOL> if metadata : <EOL> metadata . delete ( user_id , '<STR_LIT>' ) <EOL> data = self . _get_file_data ( filepath , None ) <EOL> if data : <EOL> data . delete ( user_id , '<STR_LIT>' ) <EOL> def isfile ( self , filepath ) : <EOL> """<STR_LIT>""" <EOL> metadata = self . _get_file_metadata ( filepath , None ) <EOL> return bool ( metadata ) <EOL> def listdir ( self , dir_name ) : <EOL> """<STR_LIT>""" <EOL> prefix = '<STR_LIT:%s>' % utils . vfs_construct_path ( <EOL> '<STR_LIT:/>' , self . _exploration_id , '<STR_LIT>' , dir_name ) <EOL> if not prefix . endswith ( '<STR_LIT:/>' ) : <EOL> prefix += '<STR_LIT:/>' <EOL> result = set ( ) <EOL> metadata_models = file_models . FileMetadataModel . get_undeleted ( ) <EOL> for metadata_model in metadata_models : <EOL> filepath = metadata_model . id <EOL> if filepath . startswith ( prefix ) : <EOL> result . add ( '<STR_LIT:/>' . join ( filepath . split ( '<STR_LIT:/>' ) [ <NUM_LIT:3> : ] ) ) <EOL> return sorted ( list ( result ) ) <EOL> class DiskBackedFileSystem ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , root ) : <EOL> """<STR_LIT>""" <EOL> self . _root = os . path . join ( os . getcwd ( ) , root ) <EOL> self . _exploration_id = '<STR_LIT:test>' <EOL> @ property <EOL> def exploration_id ( self ) : <EOL> return self . _exploration_id <EOL> def isfile ( self , filepath ) : <EOL> """<STR_LIT>""" <EOL> return os . path . isfile ( os . path . join ( self . _root , filepath ) ) <EOL> def get ( self , filepath , version = None , mode = '<STR_LIT:r>' ) : <EOL> """<STR_LIT>""" <EOL> content = utils . get_file_contents ( <EOL> os . path . join ( self . _root , filepath ) , raw_bytes = True , mode = mode ) <EOL> return FileStreamWithMetadata ( content , None , None ) <EOL> def commit ( self , user_id , filepath , raw_bytes ) : <EOL> raise NotImplementedError <EOL> def delete ( self , user_id , filepath ) : <EOL> raise NotImplementedError <EOL> def listdir ( self , dir_name ) : <EOL> raise NotImplementedError <EOL> class AbstractFileSystem ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , impl ) : <EOL> self . _impl = impl <EOL> @ property <EOL> def impl ( self ) : <EOL> return self . _impl <EOL> def _check_filepath ( self , filepath ) : <EOL> """<STR_LIT>""" <EOL> base_dir = utils . vfs_construct_path ( <EOL> '<STR_LIT:/>' , self . impl . exploration_id , '<STR_LIT>' ) <EOL> absolute_path = utils . vfs_construct_path ( base_dir , filepath ) <EOL> normalized_path = utils . vfs_normpath ( absolute_path ) <EOL> if not normalized_path . startswith ( base_dir ) : <EOL> raise IOError ( '<STR_LIT>' % filepath ) <EOL> def isfile ( self , filepath ) : <EOL> """<STR_LIT>""" <EOL> self . _check_filepath ( filepath ) <EOL> return self . _impl . isfile ( filepath ) <EOL> def open ( self , filepath , version = None , mode = '<STR_LIT:r>' ) : <EOL> """<STR_LIT>""" <EOL> self . _check_filepath ( filepath ) <EOL> return self . _impl . get ( filepath , version = version , mode = mode ) <EOL> def get ( self , filepath , version = None , mode = '<STR_LIT:r>' ) : <EOL> """<STR_LIT>""" <EOL> file_stream = self . open ( filepath , version = version , mode = mode ) <EOL> if file_stream is None : <EOL> raise IOError ( <EOL> '<STR_LIT>' <EOL> % ( filepath , version if version else '<STR_LIT>' ) ) <EOL> return file_stream . read ( ) <EOL> def commit ( self , user_id , filepath , raw_bytes ) : <EOL> """<STR_LIT>""" <EOL> raw_bytes = str ( raw_bytes ) <EOL> self . _check_filepath ( filepath ) <EOL> self . _impl . commit ( user_id , filepath , raw_bytes ) <EOL> def delete ( self , user_id , filepath ) : <EOL> """<STR_LIT>""" <EOL> self . _check_filepath ( filepath ) <EOL> self . _impl . delete ( user_id , filepath ) <EOL> def listdir ( self , dir_name ) : <EOL> """<STR_LIT>""" <EOL> self . _check_filepath ( dir_name ) <EOL> return self . _impl . listdir ( dir_name ) </s>
