o NZhiA@sddlZddlZddlZddlZddlZddlZddlmZddlm Z ddl Z ddl m Z m Z ddlmZmZedZGdd d e ZGd d d e ZdS) N)suppress)quote)AbstractBufferedFileAbstractFileSystem)infer_storage_optionstokenizewebhdfscs8eZdZdZeeZdZ           d=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/d0Zd@d1d2Z d3d4Z!d@d5d6Z"d7d8Z#d9d:Z$d;d<Z%Z&S)CWebHDFSa 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|jrdStjdi| | rdndd|d|d|_||_|p$i|_i|_| p,i|_|durC|dus:|dur>td||jd<||_ ||_ |durV|durUtd n |dur_||jd <|durh||jd <|rr|durrtd | |_ | |_ | 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__rU/var/www/html/lang_env/lib/python3.10/site-packages/fsspec/implementations/webhdfs.pyr-sF9      zWebHDFS.__init__cCs|jSN)rrrrr&fsidsz WebHDFS.fsidcCst|_|jr |j|j_|j|j_|jr%ddlm }|di|j |j_ |j dur@|j durBddlm}||j |j |j_ dSdSdS)Nr)HTTPKerberosAuth) HTTPBasicAuthr)requestsSessionsessionrcertrverifyrZrequests_kerberosr*rauthrrZ requests.authr+)rr*r+rrr&rs     zWebHDFS._connectgetc Ks||jt|pddd}|}||j||d<td|||j j |||||d} | j dvrqz| } | dd } | dd } Wn t tfyTYnw| d vr]t | | d vret| | d vrmt| t| | | S)Nz/=)safeopzsending %s with %s)methodrparamsdataallow_redirects)iiiiiZRemoteExceptionmessage exception)ZIllegalArgumentExceptionZUnsupportedOperationException)ZSecurityExceptionZAccessControlException)ZFileNotFoundException) _apply_proxyrrcopyupdaterupperloggerdebugr.request status_codejsonrKeyErrorPermissionErrorFileNotFoundError RuntimeErrorraise_for_status) rr5r6pathr8redirectr#rargsouterrmsgexprrr&_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 WebHDFilerU)rrJrSrTrVrWrXr#rrr&_opens !z WebHDFS._opencCs |d|d<|d|d<|S)Ntypelengthsize)lower)inforrr& _process_infos zWebHDFS._process_infocCs t|dS)NrJ)r)clsrJrrr&_strip_protocols zWebHDFS._strip_protocolcCs:t|}|dd|ddd|vr|d|d<|S)NrJprotocolusernamer)rpop)ZurlpathrMrrr&_get_kwargs_from_urlss   zWebHDFS._get_kwargs_from_urlscCs,|jd|d}|d}||d<||S)NZ GETFILESTATUSrJ FileStatusname)rQrDra)rrJrMr`rrr&r` s  z WebHDFS.infocCsr|jd|d}|dd}|D]}|||dd|d|d<q|r0t|dd d Std d |DS) NZ LISTSTATUSrhZ FileStatusesri/Z pathSuffixrjcSs|dS)Nrjr)irrr&szWebHDFS.ls..)keycss|]}|dVqdS)rjNr).0r`rrr& szWebHDFS.ls..)rQrDrarstripsorted)rrJdetailrMinfosr`rrr&lss z WebHDFS.lscCs|jd|d}|dS)z8Total numbers of files, directories and bytes under pathZGETCONTENTSUMMARYrhZContentSummaryrQrD)rrJrMrrr&content_summarys zWebHDFS.content_summarycCs^|jd|dd}d|jvr%||jd}|j|}||dS||dS)z/Checksum info of file, giving method and resultZGETFILECHECKSUMF)rJrKLocationZ FileChecksum)rQheadersr<r.r2rIrD)rrJrMlocationout2rrr&ukey s    z WebHDFS.ukeycCs|d}|dS)zGet user's home directoryZGETHOMEDIRECTORYPathrv)rrMrrr&home_directory,s  zWebHDFS.home_directorycCsB|r |jd|d}n|d}|d}|durtd|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)rQrDr)rrrMtrrr&get_delegation_token1s  zWebHDFS.get_delegation_tokencCs|jdd|d}|dS)z/Make token live longer. Returns new expiry timeZRENEWDELEGATIONTOKENputr6r"longrv)rr"rMrrr&renew_delegation_tokenBs zWebHDFS.renew_delegation_tokencCs|jdd|ddS)z Stop the token from being usefulZCANCELDELEGATIONTOKENrrNrQ)rr"rrr&cancel_delegation_tokenGszWebHDFS.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)r6rJZ permissionNr)rrJmodrrr&chmodKs z WebHDFS.chmodcCs>i}|dur ||d<|dur||d<|jdd|d|dS)zChange owning user and/or groupNownergroupSETOWNERrr6rJ)rr)rrJrrr#rrr&chownXs 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)rJr6rWNr)rrJrWrrr&set_replicationas zWebHDFS.set_replicationcKs|jdd|ddS)NZMKDIRSrrrrrJr#rrr&mkdirosz WebHDFS.mkdircCs(|dur ||r t|||dS)NF)existsFileExistsErrorr)rrJexist_okrrr&makedirsrszWebHDFS.makedirscKs|jdd||ddS)NZRENAMEr)r6rJ destinationr)rZpath1Zpath2r#rrr&mvwz WebHDFS.mvcKs|jdd||r dndddS)NDELETEdeletetruefalse)r6rJ recursiver)rrJrr#rrr&rmzs   z WebHDFS.rmcKs||dSr')rrrrr&rm_fileszWebHDFS.rm_filec Ks||_}d||dtdg}z$||d}t||Wdn1s.wY|||Wn!ty[t t  | |Wd1sUwYwWddS1sgwYdS)Nrkz.tmp.wb) openjoin_parentsecretsZ token_hexshutil copyfileobjr BaseExceptionrrGr)rZlpathZrpathr#ZlstreamZ tmp_fnameZrstreamrrr&cp_files"      "zWebHDFS.cp_filecCsJ|jrt|jr||}|S|jr#|jD] \}}|||d}q|S)N)rcallableitemsreplace)rrzkvrrr&r<s zWebHDFS._apply_proxy) r FNNNNNNFNT)r2NNT)rRNTNNFr')NN)'__name__ __module__ __qualname____doc__strtempfile gettempdirrUrdrpropertyr)rrQr[ staticmethodra classmethodrcrgr`rurwr|r~rrrrrrrrrrrrr< __classcell__rrr$r&r sd g  # -          r csJeZdZdZfddZdddZddZd d Zd d Zd dZ Z S)rZz"A file living in HDFS over webHDFSc stj||fi||}|dddur|dd|dddur+|dd|dd|_|d}|dddurQ|j|_tj |t t |_dSdS)NrXrWirUrVF) rrr=r2rfrXrJtargetosrruuiduuid4)rfsrJr#rUr$rr&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)r8ryT)rr.postrzbuffergetvaluerI)rfinalrMrrr& _upload_chunks zWebHDFile._upload_chunkcCs|j}d|jvrd\}}nd\}}d|d<|jj|||jfddi|}|j|jd}d |jvr^|jjj |d d id }| |jjd d|jfddi|}|j|jd|_ dSdS)zCreate remote file/uploada)APPENDPOST)ZCREATEPUTr overwriterKFrxwrr)ryrrN) r#r=rSrrQrJr<ryr.rrIrz)rr#r5r6rMrzr{rrr&_initiate_uploads    zWebHDFile._initiate_uploadcCst|d}t|j|}||ks||jkrdS|jjd|j|||dd}|d|jvr@|jd}|jj |j |}|j S|j S)NrZOPENF)rJoffsetr]rKrx) maxminr^rrQrJrIryr.r2r<content)rstartendrMrzr{rrr& _fetch_ranges    zWebHDFile._fetch_rangecCs|j|j|jdSr')rrrJrr(rrr&commitrzWebHDFile.commitcCs|j|jdSr')rrrJr(rrr&discardszWebHDFile.discardr) rrrrrrrrrrrrrr$r&rZs  rZ)loggingrrrrr contextlibr urllib.parserr,specrrutilsrr getLoggerr@r rZrrrr&s"