U xʹjA@sddlZddlZddlZddlZddlZddlZddlmZddlm Z ddl Z ddl m Z m Z ddlmZmZedZGdd d e ZGd d d e ZdS) N)suppress)quote)AbstractBufferedFileAbstractFileSystem)infer_storage_optionstokenizewebhdfsc seZdZdZeeZdZd=fdd Z e d d Z d d Z d>ddZ d?ddZeddZeddZeddZddZd@ddZddZdd Zd!d"ZdAd#d$Zd%d&Zd'd(Zd)d*ZdBd+d,Zd-d.Zd/d0ZdCd1d2Z d3d4Z!dDd5d6Z"d7d8Z#d9d:Z$d;d<Z%Z&S)EWebHDFSa Interface to HDFS over HTTP using the WebHDFS API. Supports also HttpFS gateways. Four auth mechanisms are supported: insecure: no auth is done, and the user is assumed to be whoever they say they are (parameter ``user``), or a predefined value such as "dr.who" if not given spnego: when kerberos authentication is enabled, auth is negotiated by requests_kerberos https://github.com/requests/requests-kerberos . This establishes a session based on existing kinit login and/or specified principal/password; parameters are passed with ``kerb_kwargs`` token: uses an existing Hadoop delegation token from another secured service. Indeed, this client can also generate such tokens when not insecure. Note that tokens expire, but can be renewed (by a previously specified user) and may allow for proxying. basic-auth: used when both parameter ``user`` and parameter ``password`` are provided. )r ZwebHDFSFNTc  s |jr dStjf| | r dndd|d|d|_||_|pDi|_i|_| pTi|_|dk r|dk sp|dk rxtd||jd<||_ ||_ |dk r|dkrtd n|dk r||jd <|dk r||jd <|r|dk rtd | |_ | |_ | d t|||_dS)a Parameters ---------- host: str Name-node address port: int Port for webHDFS kerberos: bool Whether to authenticate with kerberos for this connection token: str or None If given, use this token on every call to authenticate. A user and user-proxy may be encoded in the token and should not be also given user: str or None If given, assert the user name to connect with password: str or None If given, assert the password to use for basic auth. If password is provided, user must be provided also proxy_to: str or None If given, the user has the authority to proxy, and this value is the user in who's name actions are taken kerb_kwargs: dict Any extra arguments for HTTPKerberosAuth, see ``_ data_proxy: dict, callable or None If given, map data-node addresses. This can be necessary if the HDFS cluster is behind a proxy, running on Docker or otherwise has a mismatch between the host-names given by the name-node and the address by which to refer to them from the client. If a dict, maps host names ``host->data_proxy[host]``; if a callable, full URLs are passed, and function must conform to ``url->data_proxy(url)``. use_https: bool Whether to connect to the Name-node using HTTPS instead of HTTP session_cert: str or Tuple[str, str] or None Path to a certificate file, or tuple of (cert, key) files to use for the requests.Session session_verify: str, bool or None Path to a certificate file to use for verifying the requests.Session. kwargs Nhttpshttpz://:z /webhdfs/v1z_If passing a delegation token, must not set user or proxy_to, as these are encoded in the tokenZ delegationzQIf passing a password, the user must also beset in order to set up the basic-authz user.nameZdoaszJIf using Kerberos auth, do not specify the user, this is handled by kinit.Zwebhdfs_)_cachedsuper__init__urlkerb kerb_kwargsparsproxy ValueErroruserpassword session_certsession_verify_connectr_fsid)selfhostportZkerberostokenrrZproxy_torZ data_proxyZ use_httpsrrkwargs __class__]/var/www/html/bwave/python/venv/lib/python3.8/site-packages/fsspec/implementations/webhdfs.pyr-sB9       zWebHDFS.__init__cCs|jSN)rrr%r%r&fsidsz WebHDFS.fsidcCs|t|_|jr|j|j_|j|j_|jrFddlm }|f|j |j_ |j dk rx|j dk rxddlm}||j |j |j_ dS)Nr)HTTPKerberosAuth) HTTPBasicAuth)requestsSessionsessionrcertrverifyrZrequests_kerberosr*rauthrrZ requests.authr+)rr*r+r%r%r&rs     zWebHDFS._connectgetc Ks|dk r||nd}||jt|dd}|}||j||d<t d|||j j |||||d} | j dkrz$| } | dd } | dd } Wnttfk rYn@X| d krt| n,| d krt| n| d krt| nt| | | S)Nz/=)safeopzsending %s with %s)methodrparamsdataallow_redirects)iiiiiZRemoteExceptionmessage exception)ZIllegalArgumentExceptionZUnsupportedOperationException)ZSecurityExceptionZAccessControlException)ZFileNotFoundException)_strip_protocol _apply_proxyrrcopyupdaterupperloggerdebugr.request status_codejsonrKeyErrorPermissionErrorFileNotFoundError RuntimeErrorraise_for_status) rr5r6pathr8redirectr"rargsouterrmsgexpr%r%r&_calls:       z WebHDFS._callrbc Ks$|p|j}t|||||j|||dS)a^ Parameters ---------- path: str File location mode: str 'rb', 'wb', etc. block_size: int Client buffer size for read-ahead or write buffer autocommit: bool If False, writes to temporary file that only gets put in final location upon commit replication: int Number of copies of file on the cluster, write mode only permissions: str or int posix permissions, write mode only kwargs Returns ------- WebHDFile instance )mode block_sizetempdir autocommit replication permissions) blocksize WebHDFilerV)rrKrTrUrWrXrYr"r%r%r&_opens! z WebHDFS._opencCs |d|d<|d|d<|S)Ntypelengthsize)lower)infor%r%r& _process_infos zWebHDFS._process_infocCs t|dS)NrK)r)clsrKr%r%r&r<szWebHDFS._strip_protocolcCs:t|}|dd|ddd|kr6|d|d<|S)NrKprotocolusernamer)rpop)ZurlpathrNr%r%r&_get_kwargs_from_urlss   zWebHDFS._get_kwargs_from_urlscCs,|jd|d}|d}||d<||S)NZ GETFILESTATUSrK FileStatusname)rRrErb)rrKrNrar%r%r&ra s z WebHDFS.infocCsv|jd|d}|dd}|D](}|||dd|d|d<q"|r`t|dd d Std d |DSdS) NZ LISTSTATUSrhZ FileStatusesri/Z pathSuffixrjcSs|dS)Nrjr%)ir%r%r&zWebHDFS.ls..)keycss|]}|dVqdS)rjNr%).0rar%r%r& szWebHDFS.ls..)rRrErbrstripsorted)rrKdetailrNinfosrar%r%r&lss z WebHDFS.lscCs|jd|d}|dS)z8Total numbers of files, directories and bytes under pathZGETCONTENTSUMMARYrhZContentSummaryrRrE)rrKrNr%r%r&content_summaryszWebHDFS.content_summarycCsb|jd|dd}d|jkrJ||jd}|j|}||dS||dSdS)z/Checksum info of file, giving method and resultZGETFILECHECKSUMF)rKrLLocationZ FileChecksumN)rRheadersr=r.r2rJrE)rrKrNlocationout2r%r%r&ukeys   z WebHDFS.ukeycCs|d}|dS)zGet user's home directoryZGETHOMEDIRECTORYPathrw)rrNr%r%r&home_directory+s zWebHDFS.home_directorycCsB|r|jd|d}n |d}|d}|dkr:td|dS)zRetrieve token which can give the same authority to other uses Parameters ---------- renewer: str or None User who may use this token; if None, will be current user ZGETDELEGATIONTOKEN)renewerTokenNz1No token available for this user/security contextZ urlString)rRrEr)rrrNtr%r%r&get_delegation_token0s  zWebHDFS.get_delegation_tokencCs|jdd|d}|dS)z/Make token live longer. Returns new expiry timeZRENEWDELEGATIONTOKENputr6r!longrw)rr!rNr%r%r&renew_delegation_tokenAszWebHDFS.renew_delegation_tokencCs|jdd|ddS)z Stop the token from being usefulZCANCELDELEGATIONTOKENrrNrR)rr!r%r%r&cancel_delegation_tokenFszWebHDFS.cancel_delegation_tokencCs|jdd||ddS)aSet the permission at path Parameters ---------- path: str location to set (file or directory) mod: str or int posix epresentation or permission, give as oct string, e.g, '777' or 0o777 Z SETPERMISSIONr)r6rKZ permissionNr)rrKmodr%r%r&chmodJs z WebHDFS.chmodcCs>i}|dk r||d<|dk r$||d<|jdd|d|dS)zChange owning user and/or groupNownergroupSETOWNERrr6rK)rr)rrKrrr"r%r%r&chownWs z WebHDFS.chowncCs|jd|d|ddS)a9 Set file replication factor Parameters ---------- path: str File location (not for directories) replication: int Number of copies of file on the cluster. Should be smaller than number of data nodes; normally 3 on most systems. ZSETREPLICATIONr)rKr6rXNr)rrKrXr%r%r&set_replication`s zWebHDFS.set_replicationcKs|jdd|ddS)NZMKDIRSrrrrrKr"r%r%r&mkdirnsz WebHDFS.mkdircCs(|dkr||rt|||dS)NF)existsFileExistsErrorr)rrKexist_okr%r%r&makedirsqszWebHDFS.makedirscKs|jdd||ddS)NZRENAMEr)r6rKZ destinationr)rZpath1Zpath2r"r%r%r&mvvsz WebHDFS.mvcKs|jdd||rdndddS)NDELETEdeletetruefalse)r6rK recursiver)rrKrr"r%r%r&rmys  z WebHDFS.rmcKs||dSr')rrr%r%r&rm_fileszWebHDFS.rm_filec Ks||}d||dtdg}z4||d}t||W5QRX|||Wn4tk rt t | |W5QRXYnXW5QRXdS)Nrkz.tmp.wb) openjoinZ_parentsecretsZ token_hexshutil copyfileobjr BaseExceptionrrHr)rZlpathZrpathr"ZlstreamZ tmp_fnameZrstreamr%r%r&cp_files   zWebHDFS.cp_filecCsH|jrt|jr||}n(|jrD|jD]\}}|||d}q,|S)N)rcallableitemsreplace)rr{kvr%r%r&r=s  zWebHDFS._apply_proxy) r FNNNNNNFNT)r2NNT)rSNTNN)F)N)NN)F)F)'__name__ __module__ __qualname____doc__strtempfile gettempdirrVrdrpropertyr)rrRr\ staticmethodrb classmethodr<rgrarvrxr}rrrrrrrrrrrrrr= __classcell__r%r%r#r&r sb e  $ -         r csJeZdZdZfddZdddZddZd d Zd d Zd dZ Z S)r[z"A file living in HDFS over webHDFSc stj||f||}|dddkr6|dd|dddkrR|dd|dd|_|d}|dddkr|j|_tj |t t |_dS)NrYrXirVrWF) rrr>r2rfrYrKtargetosrruuiduuid4)rfsrKr"rVr#r%r&rs   zWebHDFile.__init__FcCs,|jjj|j|jddid}|dS)zWrite one part of a multi-block file upload Parameters ========== final: bool This is the last block, so should complete file, if self.autocommit is True. content-typeapplication/octet-stream)r8rzT)rr.postr{buffergetvaluerJ)rfinalrNr%r%r& _upload_chunks zWebHDFile._upload_chunkcCs|j}d|jkrd\}}nd\}}d|d<|jj|||jfddi|}|j|jd}d |jkr|jjj |d d id }| |jjd d|jfddi|}|j|jd|_ dS)zCreate remote file/uploada)APPENDPOST)ZCREATEPUTr overwriterLFrywrr)rzrrN) r"r>rTrrRrKr=rzr.rrJr{)rr"r5r6rNr{r|r%r%r&_initiate_uploads    zWebHDFile._initiate_uploadcCst|d}t|j|}||ks(||jkr,dS|jjd|j|||dd}|d|jkr|jd}|jj |j |}|j S|j SdS)NrrnZOPENF)rKoffsetr^rLry) maxminr_rrRrKrJrzr.r2r=content)rstartendrNr{r|r%r%r& _fetch_ranges"    zWebHDFile._fetch_rangecCs|j|j|jdSr')rrrKrr(r%r%r&commitszWebHDFile.commitcCs|j|jdSr')rrrKr(r%r%r&discardszWebHDFile.discard)F) rrrrrrrrrrrr%r%r#r&r[s r[)loggingrrrrr contextlibr urllib.parserr,specrrutilsrr getLoggerrAr r[r%r%r%r&s