OOId<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| d| rdnd|||_||_|pi|_i|_|pi|_|||td||jd<| ||jd<| ||jd <|r|td | d t||z|_ 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 Nz%{protocol}://{host}:{port}/webhdfs/v1httpshttp)protocolhostportz_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__formaturlkerb kerb_kwargsparsproxy ValueError_connectr _fsid) selfrrkerberostokenuserproxy_tor data_proxy use_httpskwargs __class__s >lib/python3.11/site-packages/fsspec/implementations/webhdfs.pyrzWebHDFS.__init__+s/\ <  F""6""":AA )5WWvDtB   &," %2  8#7  ',DIl #  %)DIk "   (DIf   (2  (4"6"66 c|jSN)r$r%s r.fsidz WebHDFS.fsidxs zr/ctj|_|jrddlm}|di|j|j_dSdS)Nr)HTTPKerberosAuthr)requestsSessionsessionrrequests_kerberosr5rauth)r%r5s r.r#zWebHDFS._connect|s_')) 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)rrcopyupdater upperloggerdebugr8request status_codejsonr"PermissionErrorFileNotFoundError RuntimeErrorKeyErrorraise_for_status) r%r>r?pathrAredirectr,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 WebHDFilerd)r%rXrbrcrerfrgr,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)NrX)r)clsrXs r._strip_protocolzWebHDFS._strip_protocols$T**622r/ct|}|dd|ddd|vr|d|d<|S)NrXrusernamer()rpop)urlpathr[s r._get_kwargs_from_urlszWebHDFS._get_kwargs_from_urlss\#G,,   D!!!   ''*--CK r/c|d|}|d}||d<||S)N GETFILESTATUSrX FileStatusname)r_rRrq)r%rXr[rps r.rpz 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).0rps r. zWebHDFS.ls..s&994$v,999999r/)r_rRrqrstripsorted)r%rXdetailr[infosrps 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_rR)r%rXr[s 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)rXrYLocation FileChecksum)r_headers _apply_proxyr8r;rWrR)r%rXr[locationout2s r.ukeyz WebHDFS.ukeysjj*jFF  $ $((Z)@AAH<##H--D  ! ! # # #99;;~. .  " " "88::n- -r/c`|d}|dS)zGet user's home directoryGETHOMEDIRECTORYPathr)r%r[s 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_rRr")r%rr[ts r.get_delegation_tokenzWebHDFS.get_delegation_token sf  3**17*CCCC**122C HHJJw  9PQQ Q~r/cf|dd|}|dS)z/Make token live longer. Returns new expiry timeRENEWDELEGATIONTOKENputr?r'longr)r%r'r[s 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_token s! *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?rX permissionNr)r%rXmods r.chmodz WebHDFS.chmod$s$ ?5t LLLLLr/cFi}|||d<|||d<|jdd|d|dS)zChange owning user and/or groupNownergroupSETOWNERrr?rX)rr)r%rXrrr,s r.chownz WebHDFS.chown1sJ  #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)rXr?rfNr)r%rXrfs r.set_replicationzWebHDFS.set_replication:s% #$u+ VVVVVr/c 6|dd|dS)NMKDIRSrrrr%rXr,s r.mkdirz WebHDFS.mkdirHs  8E 55555r/c|dur$||rt|||dS)NF)existsFileExistsErrorr)r%rXexist_oks r.makedirszWebHDFS.makedirsKsD u  T!2!2 !$'' ' 4r/c 8|dd||dS)NRENAMEr)r?rX destinationr)r%path1path2r,s r.mvz WebHDFS.mvPs" 8E5 IIIIIr/c @|dd||rdnddS)NDELETEdeletetruefalse)r?rX recursiver)r%rXrr,s r.rmz WebHDFS.rmSs<  )6ffw      r/c 0||dSr1)rrs r.rm_filezWebHDFS.rm_file[s  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 BaseExceptionrrTr)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)r!callableitemsreplace)r%rkvs r.rzWebHDFS._apply_proxylsy : 5(4:.. 5zz(++HH Z 5 ((** 5 51#++Aq!44r/)rFNNNNNF)r;NNT)r`NTNNFr1)NN)'__name__ __module__ __qualname____doc__strtempfile gettempdirrdrrpropertyr3r#r_rj staticmethodrq classmethodrtryrprrrrrrrrrrrrrrrrr __classcell__r-s@r.r r sz&c%(%''((G#H  K7K7K7K7K7K7ZXEEEF+ + + + Z\ 33[3\((( : : : :,,, . . .""" """" GGG M M MBBBB W W W666 JJJ       r/r cBeZdZdZfdZd dZdZdZdZdZ xZ S) riz"A file living in HDFS over webHDFSc Htj||fi||}|dd|dd|dd|dd|dd|_|d}|dddurQ|j|_tj |ttj |_dSdS)NrgrfirdreF) rrrKr;rwrgrXtargetosrruuiduuid4)r%fsrXr,rdr-s r.rzWebHDFile.__init__ysT,,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)rArT)rr8postrbuffergetvaluerW)r%finalr[s 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 overwriterYFrwrr)rrrN) r,rKrbrr_rXrrr8rrWr)r%r,r>r?r[rrs 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)rXoffsetrmrYr) maxminrnrr_rXrWrr8r;rcontent)r%startendr[rrs r. _fetch_rangezWebHDFile._fetch_rangesE1 $)S!! C<<5DI--3gmm 5uu     $ ${:.H7?&&tw';';H'E'EFFD< ; r/cP|j|j|jdSr1)rrrXrr2s r.commitzWebHDFile.commits"  49dk*****r/cD|j|jdSr1)rrrXr2s r.discardzWebHDFile.discards  49r/r) rrrrrr rrr r"rrs@r.ririvs,, A A A A A"KKK( +++r/ri)loggingrrrrr contextlibr urllib.parserr6specrrutilsrr getLoggerrNr rirr/r.r)s   ;;;;;;;;33333333  9 % %_____ ___D IIIII$IIIIIr/