<s> from __future__ import absolute_import <EOL> import sys <EOL> import errno <EOL> import itertools <EOL> import logging <EOL> import os <EOL> import posixpath <EOL> from boto . exception import S3ResponseError <EOL> from boto . s3 . key import Key <EOL> from boto . s3 . prefix import Prefix <EOL> from aws import s3 <EOL> from aws . s3 import translate_s3_error , s3file <EOL> from aws . s3 . s3stat import S3Stat <EOL> from hadoop . fs import normpath <EOL> DEFAULT_READ_SIZE = <NUM_LIT> * <NUM_LIT> <EOL> LOG = logging . getLogger ( __name__ ) <EOL> class S3FileSystem ( object ) : <EOL> def __init__ ( self , s3_connection ) : <EOL> self . _s3_connection = s3_connection <EOL> self . _bucket_cache = None <EOL> def _init_bucket_cache ( self ) : <EOL> if self . _bucket_cache is None : <EOL> buckets = self . _s3_connection . get_all_buckets ( ) <EOL> self . _bucket_cache = { } <EOL> for bucket in buckets : <EOL> self . _bucket_cache [ bucket . name ] = bucket <EOL> def _get_bucket ( self , name ) : <EOL> self . _init_bucket_cache ( ) <EOL> if name not in self . _bucket_cache : <EOL> self . _bucket_cache [ name ] = self . _s3_connection . get_bucket ( name ) <EOL> return self . _bucket_cache [ name ] <EOL> def _get_key ( self , path , validate = True ) : <EOL> bucket_name , key_name = s3 . parse_uri ( path ) [ : <NUM_LIT:2> ] <EOL> bucket = self . _get_bucket ( bucket_name ) <EOL> try : <EOL> return bucket . get_key ( key_name , validate = validate ) <EOL> except : <EOL> e , exc , tb = sys . exc_info ( ) <EOL> raise ValueError ( e ) <EOL> def _stats ( self , path ) : <EOL> if s3 . is_root ( path ) : <EOL> return S3Stat . for_s3_root ( ) <EOL> try : <EOL> key = self . _get_key ( path , validate = True ) <EOL> except S3ResponseError as e : <EOL> if e . status == <NUM_LIT> : <EOL> return None <EOL> else : <EOL> exc_class , exc , tb = sys . exc_info ( ) <EOL> raise exc_class , exc , tb <EOL> if key is None : <EOL> key = self . _get_key ( path , validate = False ) <EOL> return self . _stats_key ( key ) <EOL> @ staticmethod <EOL> def _stats_key ( key ) : <EOL> if key . size is not None : <EOL> is_directory_name = not key . name or key . name [ - <NUM_LIT:1> ] == '<STR_LIT:/>' <EOL> return S3Stat . from_key ( key , is_dir = is_directory_name ) <EOL> else : <EOL> key . name = S3FileSystem . _append_separator ( key . name ) <EOL> ls = key . bucket . get_all_keys ( prefix = key . name , max_keys = <NUM_LIT:1> ) <EOL> if len ( ls ) > <NUM_LIT:0> : <EOL> return S3Stat . from_key ( key , is_dir = True ) <EOL> return None <EOL> @ staticmethod <EOL> def _append_separator ( path ) : <EOL> if path and not path . endswith ( '<STR_LIT:/>' ) : <EOL> path += '<STR_LIT:/>' <EOL> return path <EOL> @ staticmethod <EOL> def _cut_separator ( path ) : <EOL> return path . endswith ( '<STR_LIT:/>' ) and path [ : - <NUM_LIT:1> ] or path <EOL> @ staticmethod <EOL> def isroot ( path ) : <EOL> return s3 . is_root ( path ) <EOL> @ staticmethod <EOL> def join ( * comp_list ) : <EOL> return s3 . join ( * comp_list ) <EOL> @ staticmethod <EOL> def normpath ( path ) : <EOL> return normpath ( path ) <EOL> @ translate_s3_error <EOL> def open ( self , path , mode = '<STR_LIT:r>' ) : <EOL> key = self . _get_key ( path , validate = True ) <EOL> if key is None : <EOL> raise IOError ( errno . ENOENT , "<STR_LIT>" % path ) <EOL> return s3file . open ( key , mode = mode ) <EOL> @ translate_s3_error <EOL> def read ( self , path , offset , length ) : <EOL> fh = self . open ( path , '<STR_LIT:r>' ) <EOL> fh . seek ( offset , os . SEEK_SET ) <EOL> return fh . read ( length ) <EOL> @ translate_s3_error <EOL> def isfile ( self , path ) : <EOL> stat = self . _stats ( path ) <EOL> if stat is None : <EOL> return False <EOL> return not stat . isDir <EOL> @ translate_s3_error <EOL> def isdir ( self , path ) : <EOL> stat = self . _stats ( path ) <EOL> if stat is None : <EOL> return False <EOL> return stat . isDir <EOL> @ translate_s3_error <EOL> def exists ( self , path ) : <EOL> return self . _stats ( path ) is not None <EOL> @ translate_s3_error <EOL> def stats ( self , path ) : <EOL> path = normpath ( path ) <EOL> stats = self . _stats ( path ) <EOL> if stats : <EOL> return stats <EOL> raise IOError ( errno . ENOENT , "<STR_LIT>" % path ) <EOL> @ translate_s3_error <EOL> def listdir_stats ( self , path , glob = None ) : <EOL> if glob is not None : <EOL> raise NotImplementedError ( "<STR_LIT>" ) <EOL> if s3 . is_root ( path ) : <EOL> self . _init_bucket_cache ( ) <EOL> return [ S3Stat . from_bucket ( b ) for b in self . _bucket_cache . values ( ) ] <EOL> bucket_name , prefix = s3 . parse_uri ( path ) [ : <NUM_LIT:2> ] <EOL> bucket = self . _get_bucket ( bucket_name ) <EOL> prefix = self . _append_separator ( prefix ) <EOL> res = [ ] <EOL> for item in bucket . list ( prefix = prefix , delimiter = '<STR_LIT:/>' ) : <EOL> if isinstance ( item , Prefix ) : <EOL> res . append ( S3Stat . from_key ( Key ( item . bucket , item . name ) , is_dir = True ) ) <EOL> else : <EOL> if item . name == prefix : <EOL> continue <EOL> res . append ( self . _stats_key ( item ) ) <EOL> return res <EOL> def listdir ( self , path , glob = None ) : <EOL> return [ s3 . parse_uri ( x . path ) [ <NUM_LIT:2> ] for x in self . listdir_stats ( path , glob ) ] <EOL> @ translate_s3_error <EOL> def rmtree ( self , path , skipTrash = False ) : <EOL> if not skipTrash : <EOL> raise NotImplementedError ( '<STR_LIT>' ) <EOL> key = self . _get_key ( path , validate = False ) <EOL> if key . exists ( ) : <EOL> to_delete = iter ( [ key ] ) <EOL> else : <EOL> to_delete = iter ( [ ] ) <EOL> if self . isdir ( path ) : <EOL> prefix = self . _append_separator ( key . name ) <EOL> keys = key . bucket . list ( prefix = prefix ) <EOL> to_delete = itertools . chain ( keys , to_delete ) <EOL> result = key . bucket . delete_keys ( to_delete ) <EOL> if result . errors : <EOL> msg = "<STR_LIT>" % ( <EOL> len ( result . errors ) , <EOL> '<STR_LIT:\n>' . join ( map ( repr , result . errors ) ) ) <EOL> LOG . error ( msg ) <EOL> raise IOError ( msg ) <EOL> @ translate_s3_error <EOL> def remove ( self , path , skip_trash = False ) : <EOL> if not skip_trash : <EOL> raise NotImplementedError ( '<STR_LIT>' ) <EOL> key = self . _get_key ( path , validate = False ) <EOL> key . bucket . delete_key ( key . name ) <EOL> def restore ( self , * args , ** kwargs ) : <EOL> raise NotImplementedError ( '<STR_LIT>' ) <EOL> @ translate_s3_error <EOL> def mkdir ( self , path , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> stats = self . _stats ( path ) <EOL> if stats : <EOL> if stats . isDir : <EOL> return None <EOL> else : <EOL> raise IOError ( errno . ENOTDIR , "<STR_LIT>" % path ) <EOL> path = self . _append_separator ( path ) <EOL> self . create ( path ) <EOL> @ translate_s3_error <EOL> def copy ( self , src , dst , recursive = False , * args , ** kwargs ) : <EOL> self . _copy ( src , dst , recursive = recursive , use_src_basename = True ) <EOL> @ translate_s3_error <EOL> def copyfile ( self , src , dst , * args , ** kwargs ) : <EOL> if self . isdir ( dst ) : <EOL> raise IOError ( errno . EINVAL , "<STR_LIT>" % dst ) <EOL> self . _copy ( src , dst , recursive = False , use_src_basename = False ) <EOL> @ translate_s3_error <EOL> def copy_remote_dir ( self , src , dst , * args , ** kwargs ) : <EOL> self . _copy ( src , dst , recursive = True , use_src_basename = False ) <EOL> def _copy ( self , src , dst , recursive , use_src_basename ) : <EOL> src_st = self . stats ( src ) <EOL> if src_st . isDir and not recursive : <EOL> return <EOL> dst = s3 . abspath ( src , dst ) <EOL> dst_st = self . _stats ( dst ) <EOL> if src_st . isDir and dst_st and not dst_st . isDir : <EOL> raise IOError ( errno . EEXIST , "<STR_LIT>" % ( dst , src ) ) <EOL> src_bucket , src_key = s3 . parse_uri ( src ) [ : <NUM_LIT:2> ] <EOL> dst_bucket , dst_key = s3 . parse_uri ( dst ) [ : <NUM_LIT:2> ] <EOL> keep_src_basename = use_src_basename and dst_st and dst_st . isDir <EOL> src_bucket = self . _get_bucket ( src_bucket ) <EOL> dst_bucket = self . _get_bucket ( dst_bucket ) <EOL> if keep_src_basename : <EOL> cut = len ( posixpath . dirname ( src_key ) ) <EOL> if cut : <EOL> cut += <NUM_LIT:1> <EOL> else : <EOL> cut = len ( src_key ) <EOL> if not src_key . endswith ( '<STR_LIT:/>' ) : <EOL> cut += <NUM_LIT:1> <EOL> for key in src_bucket . list ( prefix = src_key ) : <EOL> if not key . name . startswith ( src_key ) : <EOL> raise RuntimeError ( "<STR_LIT>" % key . name ) <EOL> dst_name = posixpath . normpath ( s3 . join ( dst_key , key . name [ cut : ] ) ) <EOL> key . copy ( dst_bucket , dst_name ) <EOL> @ translate_s3_error <EOL> def rename ( self , old , new ) : <EOL> new = s3 . abspath ( old , new ) <EOL> self . copy ( old , new , recursive = True ) <EOL> self . rmtree ( old , skipTrash = True ) <EOL> @ translate_s3_error <EOL> def rename_star ( self , old_dir , new_dir ) : <EOL> if not self . isdir ( old_dir ) : <EOL> raise IOError ( errno . ENOTDIR , "<STR_LIT>" % old_dir ) <EOL> if self . isfile ( new_dir ) : <EOL> raise IOError ( errno . ENOTDIR , "<STR_LIT>" % new_dir ) <EOL> ls = self . listdir ( old_dir ) <EOL> for entry in ls : <EOL> self . rename ( s3 . join ( old_dir , entry ) , s3 . join ( new_dir , entry ) ) <EOL> @ translate_s3_error <EOL> def create ( self , path , overwrite = False , data = None ) : <EOL> key = self . _get_key ( path , validate = False ) <EOL> key . set_contents_from_string ( data or '<STR_LIT>' , replace = overwrite ) <EOL> @ translate_s3_error <EOL> def copyFromLocal ( self , local_src , remote_dst , * args , ** kwargs ) : <EOL> local_src = self . _cut_separator ( local_src ) <EOL> remote_dst = self . _cut_separator ( remote_dst ) <EOL> def _copy_file ( src , dst ) : <EOL> key = self . _get_key ( dst , validate = False ) <EOL> fp = open ( src , '<STR_LIT:r>' ) <EOL> key . set_contents_from_file ( fp ) <EOL> if os . path . isdir ( local_src ) : <EOL> for ( local_dir , sub_dirs , files ) in os . walk ( local_src , followlinks = False ) : <EOL> remote_dir = local_dir . replace ( local_src , remote_dst ) <EOL> if not sub_dirs and not files : <EOL> self . mkdir ( remote_dir ) <EOL> else : <EOL> for file_name in files : <EOL> _copy_file ( os . path . join ( local_dir , file_name ) , os . path . join ( remote_dir , file_name ) ) <EOL> else : <EOL> file_name = os . path . split ( local_src ) [ <NUM_LIT:1> ] <EOL> if self . isdir ( remote_dst ) : <EOL> remote_file = os . path . join ( remote_dst , file_name ) <EOL> else : <EOL> remote_file = remote_dst <EOL> _copy_file ( local_src , remote_file ) <EOL> def setuser ( self , user ) : <EOL> pass </s>
