o /iD@sddlZddlZddlZddlZddlZddlZddlmZddlmZddl m Z ddl Z ddl m Z mZddlmZmZedZGd d d eZGd d d e ZdS) N)suppress)datetime)quote)AbstractBufferedFileAbstractFileSystem)infer_storage_optionstokenizewebhdfscsHeZdZdZeeZdZ           dAfdd Z e d d Z d d Z dBddZ     dCddZeddZeddZeddZddZddZddZdDdd Zd!d"Zd#d$Zd%d&ZdEd'd(Zd)d*Zd+d,Zd-d.ZdFd/d0Zd1d2Z d3d4Z!dDd5d6Z"d7d8Z#dDd9d:Z$d;d<Z%d=d>Z&d?d@Z'Z(S)GWebHDFSa 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__rX/home/kim/smarthome/.venv/lib/python3.10/site-packages/fsspec/implementations/webhdfs.pyr.sD9       zWebHDFS.__init__cCs|jSN)rr rrr'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,)r r+r,rrr'rs     zWebHDFS._connectgetc Ks|dur ||nd}||jt|dd}|}||j||d<t d|||j j |||||d} | j dvrzz| } | dd } | dd } Wn ttfy]Ynw| d vrft| | d vrnt| | d vrvt| t| | | S)Nz/=)safeopzsending %s with %s)methodrparamsdataallow_redirects)iiiiiZRemoteExceptionmessage exception)ZIllegalArgumentExceptionZUnsupportedOperationException)ZSecurityExceptionZAccessControlException)ZFileNotFoundException)_strip_protocol _apply_proxyrrcopyupdaterupperloggerdebugr/request status_codejsonrKeyErrorPermissionErrorFileNotFoundError RuntimeErrorraise_for_status) r r6r7pathr9redirectr$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 WebHDFilerW)r rLrUrVrXrYrZr$rrr'_opens !z WebHDFS._opencCs |d|d<|d|d<|S)Ntypelengthsize)lower)inforrr' _process_infos zWebHDFS._process_infocCs t|dS)NrL)r)clsrLrrr'r=s zWebHDFS._strip_protocolcCs:t|}|dd|ddd|vr|d|d<|S)NrLprotocolusernamer)rpop)ZurlpathrOrrr'_get_kwargs_from_urlss   zWebHDFS._get_kwargs_from_urlscCs,|jd|d}|d}||d<||S)NZ GETFILESTATUSrL FileStatusname)rSrFrc)r rLrOrbrrr'rb s  z WebHDFS.infocC4||}|dd}|durt|dStd)z=Return the created timestamp of a file as a datetime.datetimemodificationTimeNz5Could not retrieve creation time (modification time).rbr3r fromtimestamprJr rLrbmtimerrr'createds  zWebHDFS.createdcCrl)z>Return the modified timestamp of a file as a datetime.datetimermNrnz%Could not retrieve modification time.rorqrrr'modifieds  zWebHDFS.modifiedcKsr|jd|d}|dd}|D]}|||dd|d|d<q|r0t|dd d Std d |DS) NZ LISTSTATUSriZ FileStatusesrj/Z pathSuffixrkcSs|dS)Nrkr)irrr'(szWebHDFS.ls..)keycss|]}|dVqdS)rkNr).0rbrrr' *szWebHDFS.ls..)rSrFrcrstripsorted)r rLdetailr$rOinfosrbrrr'ls!s z WebHDFS.lscCs|jd|d}|dS)z8Total numbers of files, directories and bytes under pathZGETCONTENTSUMMARYriZContentSummaryrSrF)r rLrOrrr'content_summary,s zWebHDFS.content_summarycCs^|jd|dd}d|jvr%||jd}|j|}||dS||dS)z/Checksum info of file, giving method and resultZGETFILECHECKSUMF)rLrMLocationZ FileChecksum)rSheadersr>r/r3rKrF)r rLrOlocationout2rrr'ukey1s    z WebHDFS.ukeycCs|d}|dS)zGet user's home directoryZGETHOMEDIRECTORYPathr)r rOrrr'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)rSrFr)r rrOtrrr'get_delegation_tokenBs  zWebHDFS.get_delegation_tokencCs|jdd|d}|dS)z/Make token live longer. Returns new expiry timeZRENEWDELEGATIONTOKENputr7r#longr)r r#rOrrr'renew_delegation_tokenSs zWebHDFS.renew_delegation_tokencCs|jdd|ddS)z Stop the token from being usefulZCANCELDELEGATIONTOKENrrNrS)r r#rrr'cancel_delegation_tokenXszWebHDFS.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)r7rLZ permissionNr)r rLmodrrr'chmod\s z WebHDFS.chmodcCs>i}|dur ||d<|dur||d<|jdd|d|dS)zChange owning user and/or groupNownergroupSETOWNERrr7rL)rr)r rLrrr$rrr'chownis 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)rLr7rYNr)r rLrYrrr'set_replicationrs zWebHDFS.set_replicationcKs|jdd|ddS)NZMKDIRSrrrr rLr$rrr'mkdirsz WebHDFS.mkdircCs(|dur ||r t|||dS)NF)existsFileExistsErrorr)r rLexist_okrrr'makedirsszWebHDFS.makedirscKs|jdd||ddS)NZRENAMEr)r7rLZ destinationr)r Zpath1Zpath2r$rrr'mvz WebHDFS.mvcKs|jdd||r dndddS)NDELETEdeletetruefalse)r7rL recursiver)r rLrr$rrr'rms   z WebHDFS.rmcKs||dSr()rrrrr'rm_fileszWebHDFS.rm_filec Ks||_}d||dtdg}z$||d}t||Wdn1s.wY|||Wn!ty[t t  | |Wd1sUwYwWddS1sgwYdS)Nruz.tmp.wb) openjoinZ_parentsecretsZ token_hexshutil copyfileobjr BaseExceptionrrIr)r ZlpathZrpathr$ZlstreamZ tmp_fnameZrstreamrrr'cp_files"      "zWebHDFS.cp_filecCsJ|jrt|jr||}|S|jr#|jD] \}}|||d}q|S)N)rcallableitemsreplace)r rkvrrr'r>s zWebHDFS._apply_proxy) r FNNNNNNFNT)r3NNT)rTNTNNFr()NN))__name__ __module__ __qualname____doc__strtempfile gettempdirrWrerpropertyr*rrSr] staticmethodrc classmethodr=rhrbrsrtrrrrrrrrrrrrrrrrr> __classcell__rrr%r'r sh e  $ -          r csJeZdZdZfddZdddZddZd d Zd d Zd dZ Z S)r\z"A file living in HDFS over webHDFSc stj||fi||}|dddur|dd|dddur+|dd|dd|_|d}|dddurQ|j|_tj |t t |_dSdS)NrZrYirWrXF) rrr?r3rgrZrLtargetosrruuiduuid4)r fsrLr$rWr%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)r9rT)rr/postrbuffergetvaluerK)r finalrOrrr' _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 overwriterMFrwrr)rrrN) r$r?rUrrSrLr>rr/rrKr)r r$r6r7rOrrrrr'_initiate_uploads    zWebHDFile._initiate_uploadcCst|d}t|j|}||ks||jkrdS|jjd|j|||dd}|d|jvr@|jd}|jj |j |}|j S|j S)NrZOPENF)rLoffsetr_rMr) maxminr`rrSrLrKrr/r3r>content)r startendrOrrrrr' _fetch_ranges    zWebHDFile._fetch_rangecCs|j|j|jdSr()rrrLrr)rrr'commitrzWebHDFile.commitcCs|j|jdSr()rrrLr)rrr'discardszWebHDFile.discardr) rrrrrrrrrrrrrr%r'r\s  r\)loggingrrrrr contextlibrr urllib.parserr-specrrutilsrr getLoggerrBr r\rrrr's$