<s> """<STR_LIT>""" <EOL> from luigi . contrib . hdfs import config as hdfs_config <EOL> from luigi . contrib . hdfs import error as hdfs_error <EOL> from luigi . contrib . hdfs import abstract_client as hdfs_abstract_client <EOL> from luigi import six <EOL> import luigi . contrib . target <EOL> import logging <EOL> import datetime <EOL> import os <EOL> logger = logging . getLogger ( '<STR_LIT>' ) <EOL> class SnakebiteHdfsClient ( hdfs_abstract_client . HdfsFileSystem ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> super ( SnakebiteHdfsClient , self ) . __init__ ( ) <EOL> self . _bite = None <EOL> self . pid = - <NUM_LIT:1> <EOL> @ staticmethod <EOL> def list_path ( path ) : <EOL> if isinstance ( path , list ) or isinstance ( path , tuple ) : <EOL> return path <EOL> if isinstance ( path , six . string_types ) : <EOL> return [ path , ] <EOL> return [ str ( path ) , ] <EOL> def get_bite ( self ) : <EOL> """<STR_LIT>""" <EOL> config = hdfs_config . hdfs ( ) <EOL> if self . pid != os . getpid ( ) or not self . _bite : <EOL> client_kwargs = dict ( filter ( <EOL> lambda k_v : k_v [ <NUM_LIT:1> ] is not None and k_v [ <NUM_LIT:1> ] != '<STR_LIT>' , six . iteritems ( { <EOL> '<STR_LIT>' : config . client_version , <EOL> '<STR_LIT>' : config . effective_user , <EOL> } ) <EOL> ) ) <EOL> if config . snakebite_autoconfig : <EOL> """<STR_LIT>""" <EOL> from snakebite . client import AutoConfigClient <EOL> self . _bite = AutoConfigClient ( ** client_kwargs ) <EOL> else : <EOL> from snakebite . client import Client <EOL> self . _bite = Client ( config . namenode_host , config . namenode_port , ** client_kwargs ) <EOL> return self . _bite <EOL> def exists ( self , path ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> return self . get_bite ( ) . test ( path , exists = True ) <EOL> except Exception as err : <EOL> raise hdfs_error . HDFSCliError ( "<STR_LIT>" , - <NUM_LIT:1> , str ( err ) , repr ( err ) ) <EOL> def move ( self , path , dest ) : <EOL> """<STR_LIT>""" <EOL> parts = dest . rstrip ( '<STR_LIT:/>' ) . split ( '<STR_LIT:/>' ) <EOL> if len ( parts ) > <NUM_LIT:1> : <EOL> dir_path = '<STR_LIT:/>' . join ( parts [ <NUM_LIT:0> : - <NUM_LIT:1> ] ) <EOL> if not self . exists ( dir_path ) : <EOL> self . mkdir ( dir_path , parents = True ) <EOL> return list ( self . get_bite ( ) . rename ( self . list_path ( path ) , dest ) ) <EOL> def rename_dont_move ( self , path , dest ) : <EOL> """<STR_LIT>""" <EOL> from snakebite . errors import FileAlreadyExistsException <EOL> try : <EOL> self . get_bite ( ) . rename2 ( path , dest , overwriteDest = False ) <EOL> except FileAlreadyExistsException : <EOL> raise luigi . target . FileAlreadyExists ( ) <EOL> def remove ( self , path , recursive = True , skip_trash = False ) : <EOL> """<STR_LIT>""" <EOL> return list ( self . get_bite ( ) . delete ( self . list_path ( path ) , recurse = recursive ) ) <EOL> def chmod ( self , path , permissions , recursive = False ) : <EOL> """<STR_LIT>""" <EOL> if type ( permissions ) == str : <EOL> permissions = int ( permissions , <NUM_LIT:8> ) <EOL> return list ( self . get_bite ( ) . chmod ( self . list_path ( path ) , <EOL> permissions , recursive ) ) <EOL> def chown ( self , path , owner , group , recursive = False ) : <EOL> """<STR_LIT>""" <EOL> bite = self . get_bite ( ) <EOL> if owner : <EOL> if group : <EOL> return all ( bite . chown ( self . list_path ( path ) , "<STR_LIT>" % ( owner , group ) , <EOL> recurse = recursive ) ) <EOL> return all ( bite . chown ( self . list_path ( path ) , owner , recurse = recursive ) ) <EOL> return list ( bite . chgrp ( self . list_path ( path ) , group , recurse = recursive ) ) <EOL> def count ( self , path ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> res = self . get_bite ( ) . count ( self . list_path ( path ) ) . next ( ) <EOL> dir_count = res [ '<STR_LIT>' ] <EOL> file_count = res [ '<STR_LIT>' ] <EOL> content_size = res [ '<STR_LIT>' ] <EOL> except StopIteration : <EOL> dir_count = file_count = content_size = <NUM_LIT:0> <EOL> return { '<STR_LIT>' : content_size , '<STR_LIT>' : dir_count , <EOL> '<STR_LIT>' : file_count } <EOL> def copy ( self , path , destination ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( "<STR_LIT>" ) <EOL> def put ( self , local_path , destination ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( "<STR_LIT>" ) <EOL> def get ( self , path , local_destination ) : <EOL> """<STR_LIT>""" <EOL> return list ( self . get_bite ( ) . copyToLocal ( self . list_path ( path ) , <EOL> local_destination ) ) <EOL> def mkdir ( self , path , parents = True , mode = <NUM_LIT> , raise_if_exists = False ) : <EOL> """<STR_LIT>""" <EOL> result = list ( self . get_bite ( ) . mkdir ( self . list_path ( path ) , <EOL> create_parent = parents , mode = mode ) ) <EOL> if raise_if_exists and "<STR_LIT>" in result [ <NUM_LIT:0> ] . get ( '<STR_LIT:error>' , '<STR_LIT>' ) : <EOL> raise luigi . target . FileAlreadyExists ( "<STR_LIT>" % ( path , ) ) <EOL> return result <EOL> def listdir ( self , path , ignore_directories = False , ignore_files = False , <EOL> include_size = False , include_type = False , include_time = False , <EOL> recursive = False ) : <EOL> """<STR_LIT>""" <EOL> bite = self . get_bite ( ) <EOL> for entry in bite . ls ( self . list_path ( path ) , recurse = recursive ) : <EOL> if ignore_directories and entry [ '<STR_LIT>' ] == '<STR_LIT:d>' : <EOL> continue <EOL> if ignore_files and entry [ '<STR_LIT>' ] == '<STR_LIT:f>' : <EOL> continue <EOL> rval = [ entry [ '<STR_LIT:path>' ] , ] <EOL> if include_size : <EOL> rval . append ( entry [ '<STR_LIT>' ] ) <EOL> if include_type : <EOL> rval . append ( entry [ '<STR_LIT>' ] ) <EOL> if include_time : <EOL> rval . append ( datetime . datetime . fromtimestamp ( entry [ '<STR_LIT>' ] / <NUM_LIT:1000> ) ) <EOL> if len ( rval ) > <NUM_LIT:1> : <EOL> yield tuple ( rval ) <EOL> else : <EOL> yield rval [ <NUM_LIT:0> ] <EOL> def touchz ( self , path ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( "<STR_LIT>" ) </s>
