<s> from posixpath import join as pjoin <EOL> import posixpath <EOL> import six <EOL> from ibis . config import options <EOL> from ibis . util import implements <EOL> import ibis . common as com <EOL> class HDFSError ( com . IbisError ) : <EOL> pass <EOL> class HDFS ( object ) : <EOL> """<STR_LIT>""" <EOL> def log ( self , message ) : <EOL> print ( message ) <EOL> def exists ( self , path ) : <EOL> raise NotImplementedError <EOL> def status ( self , path ) : <EOL> raise NotImplementedError <EOL> def chmod ( self , hdfs_path , permissions ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> def chown ( self , hdfs_path , owner = None , group = None ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> def head ( self , hdfs_path , nbytes = <NUM_LIT> , offset = <NUM_LIT:0> ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> def get ( self , hdfs_path , local_path = '<STR_LIT:.>' , overwrite = False ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> def put ( self , hdfs_path , resource , overwrite = False , verbose = None , <EOL> ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> def put_tarfile ( self , hdfs_path , local_path , compression = '<STR_LIT>' , <EOL> verbose = None , overwrite = False ) : <EOL> """<STR_LIT>""" <EOL> import tarfile <EOL> modes = { <EOL> None : '<STR_LIT:r>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> if compression not in modes : <EOL> raise ValueError ( '<STR_LIT>' <EOL> . format ( compression ) ) <EOL> mode = modes [ compression ] <EOL> tf = tarfile . open ( local_path , mode = mode ) <EOL> for info in tf : <EOL> if not info . isfile ( ) : <EOL> continue <EOL> buf = tf . extractfile ( info ) <EOL> abspath = pjoin ( hdfs_path , info . path ) <EOL> self . put ( abspath , buf , verbose = verbose , overwrite = overwrite ) <EOL> def put_zipfile ( self , hdfs_path , local_path ) : <EOL> raise NotImplementedError <EOL> def write ( self , hdfs_path , buf , overwrite = False , blocksize = None , <EOL> replication = None , buffersize = None ) : <EOL> raise NotImplementedError <EOL> def mkdir ( self , path ) : <EOL> pass <EOL> def ls ( self , hdfs_path , status = False ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> def size ( self , hdfs_path ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> def tail ( self , hdfs_path , nbytes = <NUM_LIT> ) : <EOL> raise NotImplementedError <EOL> def mv ( self , hdfs_path_src , hdfs_path_dest , overwrite = True ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> def cp ( self , hdfs_path_src , hdfs_path_dest ) : <EOL> raise NotImplementedError <EOL> def rm ( self , path ) : <EOL> """<STR_LIT>""" <EOL> return self . delete ( path ) <EOL> def rmdir ( self , path ) : <EOL> """<STR_LIT>""" <EOL> self . client . delete ( path , recursive = True ) <EOL> def _find_any_file ( self , hdfs_dir ) : <EOL> contents = self . ls ( hdfs_dir , status = True ) <EOL> def valid_filename ( name ) : <EOL> head , tail = posixpath . split ( name ) <EOL> tail = tail . lower ( ) <EOL> return ( not tail . endswith ( '<STR_LIT>' ) and <EOL> not tail . endswith ( '<STR_LIT>' ) and <EOL> not tail . startswith ( '<STR_LIT:_>' ) and <EOL> not tail . startswith ( '<STR_LIT:.>' ) ) <EOL> for filename , meta in contents : <EOL> if meta [ '<STR_LIT:type>' ] . lower ( ) == '<STR_LIT:file>' and valid_filename ( filename ) : <EOL> return filename <EOL> raise com . IbisError ( '<STR_LIT>' ) <EOL> class WebHDFS ( HDFS ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , client ) : <EOL> self . client = client <EOL> @ property <EOL> def protocol ( self ) : <EOL> return '<STR_LIT>' <EOL> def status ( self , path ) : <EOL> """<STR_LIT>""" <EOL> return self . client . status ( path ) <EOL> @ implements ( HDFS . chmod ) <EOL> def chmod ( self , path , permissions ) : <EOL> self . client . set_permission ( path , permissions ) <EOL> @ implements ( HDFS . chown ) <EOL> def chown ( self , path , owner = None , group = None ) : <EOL> self . client . set_owner ( path , owner , group ) <EOL> @ implements ( HDFS . exists ) <EOL> def exists ( self , path ) : <EOL> return not self . client . status ( path , strict = False ) is None <EOL> @ implements ( HDFS . ls ) <EOL> def ls ( self , hdfs_path , status = False ) : <EOL> return self . client . list ( hdfs_path , status = status ) <EOL> @ implements ( HDFS . mkdir ) <EOL> def mkdir ( self , dir_path ) : <EOL> self . client . makedirs ( dir_path ) <EOL> @ implements ( HDFS . size ) <EOL> def size ( self , hdfs_path ) : <EOL> return self . client . content ( hdfs_path ) [ '<STR_LIT>' ] <EOL> @ implements ( HDFS . mv ) <EOL> def mv ( self , hdfs_path_src , hdfs_path_dest , overwrite = True ) : <EOL> if overwrite and self . exists ( hdfs_path_dest ) : <EOL> if self . status ( hdfs_path_dest ) [ '<STR_LIT:type>' ] == '<STR_LIT>' : <EOL> self . rm ( hdfs_path_dest ) <EOL> return self . client . rename ( hdfs_path_src , hdfs_path_dest ) <EOL> def delete ( self , hdfs_path , recursive = False ) : <EOL> """<STR_LIT>""" <EOL> return self . client . delete ( hdfs_path , recursive = recursive ) <EOL> @ implements ( HDFS . head ) <EOL> def head ( self , hdfs_path , nbytes = <NUM_LIT> , offset = <NUM_LIT:0> ) : <EOL> _reader = self . client . read ( hdfs_path , offset = offset , length = nbytes ) <EOL> with _reader as reader : <EOL> return reader . read ( ) <EOL> @ implements ( HDFS . put ) <EOL> def put ( self , hdfs_path , resource , overwrite = False , verbose = None , <EOL> ** kwargs ) : <EOL> verbose = verbose or options . verbose <EOL> if isinstance ( resource , six . string_types ) : <EOL> return self . client . upload ( hdfs_path , resource , overwrite = overwrite , <EOL> ** kwargs ) <EOL> else : <EOL> hdfs_path = self . client . resolve ( hdfs_path ) <EOL> self . client . write ( hdfs_path , data = resource , overwrite = overwrite , <EOL> ** kwargs ) <EOL> return hdfs_path <EOL> @ implements ( HDFS . get ) <EOL> def get ( self , hdfs_path , local_path , overwrite = False , verbose = None , <EOL> ** kwargs ) : <EOL> verbose = verbose or options . verbose <EOL> return self . client . download ( hdfs_path , local_path , overwrite = overwrite , <EOL> ** kwargs ) </s>
