̒ge;ddlZddlZddlZddlZddlZddlZddlmZddlm Z ddl Z ddl m Z m Z ddlmZmZejdZGdd e ZGd d e ZdS) N)suppress)quote)AbstractBufferedFileAbstractFileSystem)infer_storage_optionstokenizewebhdfscVeZdZdZeejZdZ d#fd Z e dZ dZ d$d Z d%d ZedZedZedZdZd&dZdZdZdZd'dZdZdZdZd(dZdZdZd&dZ dZ!d&dZ"d Z#d!Z$d"Z%xZ&S))WebHDFSa~ Interface to HDFS over HTTP using the WebHDFS API. Supports also HttpFS gateways. Three 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. )r webHDFSFNc  |jrdStjd i| | rdndd|d|d|_||_|pi|_i|_|pi|_|||td||jd<| ||jd <| ||jd <|r|td | 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 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 kwargs Nhttpshttpz://:z /webhdfs/v1z_If passing a delegation token, must not set user or proxy_to, as these are encoded in the token delegationz user.namedoaszJIf using Kerberos auth, do not specify the user, this is handled by kinit.webhdfs_) _cachedsuper__init__urlkerb kerb_kwargsparsproxy ValueError_connectr _fsid) selfhostportkerberostokenuserproxy_tor data_proxy use_httpskwargs __class__s >lib/python3.11/site-packages/fsspec/implementations/webhdfs.pyrzWebHDFS.__init__+s,\ <  F""6"""!*6ggSS4SS$SSS &," %2  8#7  ',DIl #  %)DIk "   (DIf   (2  6t 4 466 c|jSN)r!r"s r-fsidz WebHDFS.fsidvs zr.ctj|_|jrddlm}|di|j|j_dSdS)Nr)HTTPKerberosAuthr)requestsSessionsessionrrequests_kerberosr4rauth)r"r4s r-r zWebHDFS._connectzs_')) 9 E : : : : : : 0 0 D D43C D DDL    E Er.getTc |jt|pdz}|}||j||d<t d|||j |||||} | j dvr | } | dd} | dd} | d vrt| | d vrt| | d vrt| t| #tt f$rYnwxYw| | S) Nopzsending %s with %s)methodrparamsdataallow_redirects)iiiiiRemoteExceptionmessage exception)IllegalArgumentExceptionUnsupportedOperationException)SecurityExceptionAccessControlException)FileNotFoundException)rrcopyupdaterupperloggerdebugr7request status_codejsonrPermissionErrorFileNotFoundError RuntimeErrorKeyErrorraise_for_status) r"r=r>pathr@redirectr+rargsouterrmsgexps r-_callz WebHDFS._callsthtzr***{{}} DIXXZZT  )3777l""<<>>$ #   ?7 7 7 ,hhjj+,Y7+,[9WWW$S//)KKK)#...555+C000&s+++)      s60D..EErbc L|p|j}t|||||j|||S)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 WebHDFilerc)r"rWrarbrdrerfr+s r-_openz WebHDFS._opensBB 14>   !L!##    r.cV|d|d<|d|d<|S)Ntypelengthsize)lower)infos r- _process_infozWebHDFS._process_infos-F|))++V H~V  r.c,t|dS)NrW)r)clsrWs r-_strip_protocolzWebHDFS._strip_protocols$T**622r.ct|}|dd|ddd|vr|d|d<|S)NrWprotocolusernamer')rpop)urlpathrZs r-_get_kwargs_from_urlszWebHDFS._get_kwargs_from_urlss\#G,,   D!!!   ''*--CK r.c|d|}|d}||d<||S)N GETFILESTATUSrW FileStatusname)r^rQrp)r"rWrZros r-roz WebHDFS.infosFjjtj44xxzz,'V !!$'''r.cF|d|}|dd}|D];}|||ddz|dz|d<<|rt |d St d |DS) N LISTSTATUSr| FileStatusesr}/ pathSuffixr~c|dS)Nr~r)is r-zWebHDFS.ls..s qyr.)keyc3&K|] }|dV dS)r~Nr).0ros r- zWebHDFS.ls..s&994$v,999999r.)r^rQrprstripsorted)r"rWdetailrZinfosros r-lsz WebHDFS.lssjjDj11 >*<8 G GD   t $ $ $;;s++c1D4FFDLL  :%%8%8999 999599999 9r.cd|d|}|dS)z8Total numbers of files, directories and bytes under pathGETCONTENTSUMMARYr|ContentSummaryr^rQ)r"rWrZs r-content_summaryzWebHDFS.content_summarys,jj,4j88xxzz*++r.cp|d|d}d|jvrh||jd}|j|}||dS||dS)z/Checksum info of file, giving method and resultGETFILECHECKSUMF)rWrXLocation FileChecksum)r^headers _apply_proxyr7r:rVrQ)r"rWrZlocationout2s r-ukeyz WebHDFS.ukeysjj*jFF  $ $((Z)@AAH<##H--D  ! ! # # #99;;~. .  " " "88::n- -r.c`|d}|dS)zGet user's home directoryGETHOMEDIRECTORYPathr)r"rZs r-home_directoryzWebHDFS.home_directorys'jj+,,xxzz&!!r.c|r|d|}n|d}|d}|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 GETDELEGATIONTOKEN)renewerTokenNz1No token available for this user/security context urlString)r^rQr)r"rrZts r-get_delegation_tokenzWebHDFS.get_delegation_tokensf  3**17*CCCC**122C HHJJw  9PQQ Q~r.cf|dd|}|dS)z/Make token live longer. Returns new expiry timeRENEWDELEGATIONTOKENputr>r&longr)r"r&rZs r-renew_delegation_tokenzWebHDFS.renew_delegation_tokens-jj/UjKKxxzz&!!r.c6|dd|dS)z Stop the token from being usefulCANCELDELEGATIONTOKENrrNr^)r"r&s r-cancel_delegation_tokenzWebHDFS.cancel_delegation_tokens! *5 FFFFFr.c8|dd||dS)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 SETPERMISSIONr)r>rW permissionNr)r"rWmods r-chmodz WebHDFS.chmod"s$ ?5t LLLLLr.cFi}|||d<|||d<|jdd|d|dS)zChange owning user and/or groupNownergroupSETOWNERrr>rW)rr)r"rWrrr+s r-chownz WebHDFS.chown/sJ  #F7O  #F7O Ae$AA&AAAAAr.c8|d|d|dS)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. SETREPLICATIONr)rWr>reNr)r"rWres r-set_replicationzWebHDFS.set_replication8s% #$u+ VVVVVr.c 6|dd|dS)NMKDIRSrrrr"rWr+s r-mkdirz WebHDFS.mkdirFs  8E 55555r.c|dur$||rt|||dS)NF)existsFileExistsErrorr)r"rWexist_oks r-makedirszWebHDFS.makedirsIsD u  T!2!2 !$'' ' 4r.c 8|dd||dS)NRENAMEr)r>rW destinationr)r"path1path2r+s r-mvz WebHDFS.mvNs" 8E5 IIIIIr.c @|dd||rdnddS)NDELETEdeletetruefalse)r>rW recursiver)r"rWrr+s r-rmz WebHDFS.rmQs<  )6ffw      r.c 0||dSr0)rrs r-rm_filezWebHDFS.rm_fileYs  r.c 8||5}d||dtjdg} ||d5}t j||dddn #1swxYwY|||nP#t$rCtt5| |dddn #1swxYwYwxYw ddddS#1swxYwYdS)Nrz.tmp.wb) openjoin_parentsecrets token_hexshutil copyfileobjr BaseExceptionrrSr)r"lpathrpathr+lstream tmp_fnamerstreams r-cp_filezWebHDFS.cp_file\s YYu   $,,u"5"57Vw?PQS?T?T7V7V!WXXI YYy$//97&w888999999999999999 5))))    /00''GGI&&&''''''''''''''' *                  srADB1-B B1B B1B B10D1C>C1% C>1C5 5C>8C5 9C>>DDDc|jr*t|jr||}n=|jr6|jD]\}}|||d}|S)N)rcallableitemsreplace)r"rkvs r-rzWebHDFS._apply_proxyjsy : 5(4:.. 5zz(++HH Z 5 ((** 5 51#++Aq!44r.)rFNNNNNF)r:NNT)r_NTNNFr0)NN)'__name__ __module__ __qualname____doc__strtempfile gettempdirrcrurpropertyr2r r^ri staticmethodrp classmethodrsryrorrrrrrrrrrrrrrrrr __classcell__r,s@r-r r sz&c%(%''((G#H  I7I7I7I7I7I7VXEEEF+ + + + Z\ 33[3\((( : : : :,,, . . .""" """" GGG M M MBBBB W W W666 JJJ       r.r cBeZdZdZfdZd dZdZdZdZdZ xZ S) rhz"A file living in HDFS over webHDFSc Htj||fi||}|dd|dd|dd|dd|dd|_|d}|dddurQ|j|_tj |ttj |_dSdS)NrfreircrdF) rrrJr:rwrfrWtargetosrruuiduuid4)r"fsrWr+rcr,s r-rzWebHDFile.__init__wsT,,V,,, ::mT * * 2 JJ}d + + + ::mT * * 2 JJ}d + + +!::mS99**Y'' ::lE * *e 3 3)DK Wc$*,,.?.?@@DIII 4 3r.Fc|jj|j|jddi}|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)r@rT)rr7postrbuffergetvaluerV)r"finalrZs r- _upload_chunkzWebHDFile._upload_chunksZgo"" M%%''#%?@#   tr.c|j}d|jvrd\}}n d\}}d|d<|jj|||jfddi|}|j|jd}d |jvr|jj |d d i }| |jjd d|jfddi|}|j|jd|_ dSdS)zCreate remote file/uploada)APPENDPOST)CREATEPUTr overwriterXFrwrr)rrrN) r+rJrarr^rWrrr7rrVr)r"r+r=r>rZrrs r-_initiate_uploadzWebHDFile._initiate_uploads!!## $)  )JB(JB"(F; dgmB LLELVLL7'' J(?@@ $)  7?&&>3M"N'D  ! ! # # # 47=649WWuWPVWWD G00j1IJJDMMM  r.ct|d}t|j|}||ks ||jkrdS|jd|j|||z d}|d|jvrK|jd}|jj |j |}|j S|j S)Nrr.OPENF)rWoffsetrlrXr) maxminrmrr^rWrVrr7r:rcontent)r"startendrZrrs r- _fetch_rangezWebHDFile._fetch_rangesE1 $)S!! C<<5DI--3gmm 5uu     $ ${:.H7?&&tw';';H'E'EFFD< ; r.cP|j|j|jdSr0)rrrWrr1s r-commitzWebHDFile.commits"  49dk*****r.cD|j|jdSr0)rrrWr1s r-discardzWebHDFile.discards  49r.r) rrrrrr rrr r"rrs@r-rhrhts,, A A A A A"KKK( +++r.rh)loggingrrrrr contextlibr urllib.parserr5specrrutilsrr getLoggerrMr rhrr.r-r)s   ;;;;;;;;33333333  9 % %]]]]] ]]]@ IIIII$IIIIIr.