"h;ddlZddlZddlZddlZddlZddlZddlmZddlm Z ddl Z ddl m Z m Z ddlmZmZej"dZGdd e ZGd d e Zy) N)suppress)quote)AbstractBufferedFileAbstractFileSystem)infer_storage_optionstokenizewebhdfsc<eZdZdZeej ZdZ dfd Z e dZ dZ ddZ ddZedZed Zed Zd Zd d Zd ZdZdZd!dZdZdZdZd"dZdZdZd dZ dZ!d dZ"dZ#dZ$dZ%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 webHDFSc  |jryt |d i| | rdndd|d|d|_||_|xsi|_i|_|xsi|_||| td||j d<|||j d <|||j d <|r | td |jd t|||_ y)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 [/var/www/html/sandstorm/venv/lib/python3.12/site-packages/fsspec/implementations/webhdfs.pyrzWebHDFS.__init__+s\ <<  "6"!*g7s4&${S &," %2  8#7  ',DIIl #  %)DIIk "   (DIIf  (2  t 456 c|jSN)r r!s r,fsidz WebHDFS.fsidvs zzr-ctj|_|jr(ddlm}|di|j |j_yy)Nr)HTTPKerberosAuthr)requestsSessionsessionrrequests_kerberosr3rauth)r!r3s r,rzWebHDFS._connectzs;'') 99 : 0 D43C3C DDLL  r-c \|jt|xsdz}|j}|j|j|j |d<t jd|||jj|j ||||} | jdvrY | j} | dd} | dd} | d vr t| | d vr t| | d vr t| t| | j#| S#tt f$rY#wxYw) Nopzsending %s with %s)methodrparamsdataallow_redirects)iiiiiRemoteExceptionmessage exception)IllegalArgumentExceptionUnsupportedOperationException)SecurityExceptionAccessControlException)FileNotFoundException)rrcopyupdaterupperloggerdebugr6request status_codejsonrPermissionErrorFileNotFoundError RuntimeErrorKeyErrorraise_for_status) r!r;r<pathr>redirectr*rargsouterrmsgexps r,_callz WebHDFS._calls3hhtzr**{{} DIIXXZT  )37ll""<<>$ #  ??7 7 ,hhj+,Y7+,[9WW$S/)KK)#..55+C00&s++  )  s/ DD+*D+c \|xs |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 WebHDFiler`)r!rUr^r_rarbrcr*s r,_openz WebHDFS._opens<B 14>>   !LL!##  r-cB|dj|d<|d|d<|S)Ntypelengthsize)lower)infos r, _process_infozWebHDFS._process_infos*F|))+V H~V  r-ct|dS)NrU)r)clsrUs r,_strip_protocolzWebHDFS._strip_protocols$T*622r-ct|}|jdd|jddd|vr|jd|d<|S)NrUprotocolusernamer&)rpop)urlpathrXs r,_get_kwargs_from_urlszWebHDFS._get_kwargs_from_urlssG#G,   D!  ''*-CK r-cz|jd|}|jd}||d<|j|S)N GETFILESTATUSrU FileStatusname)r\rOrm)r!rUrXrls r,rlz WebHDFS.infos>jjtj4xxz,'V !!$''r-c|jd|}|jdd}|D]0}|j||jddz|dz|d<2|rt |d St d |DS) N LISTSTATUSry FileStatusesrz/ pathSuffixr{c |dS)Nr{r)is r,zWebHDFS.ls..s qyr-)keyc3&K|] }|d yw)r{Nr).0rls r, zWebHDFS.ls..s94$v,9s)r\rOrmrstripsorted)r!rUdetailrXinfosrls r,lsz WebHDFS.lssjjDj1 >*<8 GD   t $;;s+c1D4FFDL G %%89 99599 9r-cN|jd|}|jdS)z8Total numbers of files, directories and bytes under pathGETCONTENTSUMMARYryContentSummaryr\rO)r!rUrXs r,content_summaryzWebHDFS.content_summarys'jj,4j8xxz*++r-cD|jd|d}d|jvr\|j|jd}|jj |}|j |j dS|j |j dS)z/Checksum info of file, giving method and resultGETFILECHECKSUMF)rUrVLocation FileChecksum)r\headers _apply_proxyr6getrTrO)r!rUrXlocationout2s r,ukeyz WebHDFS.ukeysjj*jF  $((Z)@AH<<##H-D  ! ! #99;~. .  "88:n- -r-cJ|jd}|jdS)zGet user's home directoryGETHOMEDIRECTORYPathr)r!rXs r,home_directoryzWebHDFS.home_directorys"jj+,xxz&!!r-c|r|jd|}n|jd}|jd}| 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)renewerTokenz1No token available for this user/security context urlString)r\rOr)r!rrXts r,get_delegation_tokenzWebHDFS.get_delegation_tokensT **17*CC**12C HHJw  9PQ Q~r-cP|jdd|}|jdS)z/Make token live longer. Returns new expiry timeRENEWDELEGATIONTOKENputr<r%longr)r!r%rXs r,renew_delegation_tokenzWebHDFS.renew_delegation_tokens(jj/UjKxxz&!!r-c,|jdd|y)z Stop the token from being usefulCANCELDELEGATIONTOKENrrNr\)r!r%s r,cancel_delegation_tokenzWebHDFS.cancel_delegation_tokens *5 Fr-c.|jdd||y)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<rU permissionNr)r!rUmods r,chmodz WebHDFS.chmod"s ?5t Lr-cNi}|||d<|||d<|jdd|d|y)zChange owning user and/or groupNownergrouprr<rU)SETOWNERr)r!rUrrr*s r,chownz WebHDFS.chown/s=  #F7O  #F7O Ae$A&Ar-c.|jd|d|y)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)rUr<rbNr)r!rUrbs r,set_replicationzWebHDFS.set_replication8s #$u+ Vr-c ,|jdd|y)NMKDIRSrrrr!rUr*s r,mkdirz WebHDFS.mkdirFs 8E 5r-cf|dur|j|r t||j|y)NF)existsFileExistsErrorr)r!rUexist_oks r,makedirszWebHDFS.makedirsIs, u T!2!$' ' 4r-c .|jdd||y)NRENAMEr)r<rU destinationr)r!path1path2r*s r,mvz WebHDFS.mvNs 8E5 Ir-c B|jdd||rdydy)NDELETEdeletetruefalse)r<rU recursiver)r!rUrr*s r,rmz WebHDFS.rmQs0  )f  07  r-c &|j|yr/)rrs r,rm_filezWebHDFS.rm_fileYs   r-c |j|5}dj|j|dtjdg} |j|d5}t j ||ddd|j|| dddy#1swY%xYw#t$r7tt5|j|ddd#1swYxYwwxYw#1swYyxYw)Nrz.tmp.wb) openjoin_parentsecrets token_hexshutil copyfileobjr BaseExceptionrrQr)r!lpathrpathr*lstream tmp_fnamerstreams r,cp_filezWebHDFS.cp_file\s YYu  $,,u"5w?P?PQS?T>U7V!WXI YYy$/97&&w89 5)   99! /0'GGI&''   sM9C( B%B5B%B" B%%C%=C C%C! C%%C((C1c|jr(t|jr|j|}|S|jr5|jjD]\}}|j||d}|S)N)rcallableitemsreplace)r!rkvs r,rzWebHDFS._apply_proxyjsk ::(4::.zz(+H  ZZ ((* 51#++Aq!4 5r-)iFNNNNNF)rNNT)rbNTNNFr/)NN)'__name__ __module__ __qualname____doc__strtempfile gettempdirr`rrrpropertyr1rr\rf staticmethodrm classmethodrprvrlrrrrrrrrrrrrrrrrr __classcell__r+s@r,r r s&%(%%'(G#H  I7VEF+ Z 33( :, ." "" G MB W6 J  r-r cBeZdZdZfdZddZdZdZdZdZ xZ S) rez"A file living in HDFS over webHDFSc t|||fi||j}|jdd|j dd|jdd|j dd|j dd|_|j d}|j dddurR|j |_tj j|ttj|_yy)Nrcrbir`raF) rrrHrrtrcrUtargetosrruuiduuid4)r!fsrUr*r`r+s r,rzWebHDFile.__init__ws T,V, ::mT * 2 JJ}d + ::mT * 2 JJ}d +!::mS9**Y' ::lE *e 3))DK Wc$**,.?@DI 4r-c|jjj|j|jj ddi}|j y)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)rr6postrbuffergetvaluerT)r!finalrXs r, _upload_chunkzWebHDFile._upload_chunksRggoo"" MM%%'#%?@#  r-cZ|jj}d|jvrd\}}n d\}}d|d<|jj|||j fddi|}|jj |jd}d |jvr|jjj|d d i }|j|jjd d|j fddi|}|jj |jd|_ yy)zCreate remote file/uploada)APPENDPOST)CREATEPUTr overwriterVFrwrr)rrrN) r*rHr^rr\rUrrr6rrTr)r!r*r;r<rXrrs r,_initiate_uploadzWebHDFile._initiate_uploads !!# $)) )JB(JB"(F; dggmmB LELVL77'' J(?@ $)) 77??&&>3M"N'D  ! ! # 477==6499WuWPVWD GG00j1IJDM r-ct|d}t|j|}||k\s||jk\ry|jj d|j |||z d}|j d|jvrY|jd}|jjj|jj|}|jS|jS)Nrr-OPENF)rUoffsetrirVr) maxminrjrr\rUrTrr6rrcontent)r!startendrXrrs r, _fetch_rangezWebHDFile._fetch_rangesE1 $))S! C<5DII-ggmm 5uu    ${{:.H77??&&tww';';H'EFD<< ;; r-cd|jj|j|jyr/)rrrUrr0s r,commitzWebHDFile.commits  499dkk*r-cN|jj|jyr/)rrrUr0s r,discardzWebHDFile.discards  499r-r) rrrrrr rrrr!rrs@r,rerets&, A"K( +r-re)loggingrrrrr contextlibr urllib.parserr4specrrutilsrr getLoggerrKr rerr-r,r(sY   ;3   9 %] ]@ I$Ir-