<s> """<STR_LIT>""" <EOL> from . . utils . exceptions import ETLInputError <EOL> from . . utils . helpers import parse_path <EOL> from . s3_path import S3Path <EOL> from . utils import read_from_s3 <EOL> from . utils import upload_to_s3 <EOL> DEFAULT_FILE_NAME = '<STR_LIT:file>' <EOL> class S3File ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , path = None , text = None , s3_path = None ) : <EOL> """<STR_LIT>""" <EOL> if path or text : <EOL> assert ( path and not text ) or ( text and not path ) , '<STR_LIT>' <EOL> self . _path = parse_path ( path ) <EOL> self . _text = text <EOL> self . _s3_path = s3_path <EOL> def upload_to_s3 ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _s3_path : <EOL> if self . _path or self . _text : <EOL> upload_to_s3 ( self . _s3_path , self . _path , self . _text ) <EOL> else : <EOL> raise ETLInputError ( '<STR_LIT>' ) <EOL> @ property <EOL> def text ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _text : <EOL> return self . _text <EOL> elif self . _path : <EOL> with open ( self . _path , '<STR_LIT:r>' ) as f : <EOL> return f . read ( ) <EOL> return read_from_s3 ( self . _s3_path ) <EOL> @ property <EOL> def file_name ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _path : <EOL> return self . _path . split ( '<STR_LIT:/>' ) . pop ( ) <EOL> else : <EOL> return DEFAULT_FILE_NAME <EOL> @ property <EOL> def s3_path ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _s3_path <EOL> @ s3_path . setter <EOL> def s3_path ( self , s3_path ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( s3_path , S3Path ) : <EOL> raise ETLInputError ( '<STR_LIT>' ) <EOL> self . _s3_path = S3Path ( <EOL> key = s3_path . key , <EOL> bucket = s3_path . bucket , <EOL> is_directory = s3_path . is_directory , <EOL> ) <EOL> if s3_path . is_directory : <EOL> self . _s3_path . append ( self . file_name ) </s>
