U 9%e<@sddlZddlZddlZddlZddlZddlZddlmZddlm Z ddl Z ddl m Z m Z ddlmZmZedZGdd d e ZGd d d e ZdS) N)suppress)quote)AbstractBufferedFileAbstractFileSystem)infer_storage_optionstokenizewebhdfscseZdZdZeeZdZd=fdd Z e dd Z d d Z d>ddZ d?ddZeddZeddZeddZddZd@ddZddZdd Zd!d"ZdAd#d$Zd%d&Zd'd(Zd)d*ZdBd+d,Zd-d.Zd/d0ZdCd1d2Z d3d4Z!dDd5d6Z"d7d8Z#d9d:Z$d;d<Z%Z&S)EWebHDFSa~ 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 ZwebHDFSFNc  s|jr dStjf| dj| r$dnd||d|_||_|p>i|_i|_|pNi|_|dk r||dk sj|dk rrt d||jd<|dk r||jd<|dk r||jd <|r|dk rt d | 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 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 tokenZ delegationz user.nameZdoaszJIf using Kerberos auth, do not specify the user, this is handled by kinit.Zwebhdfs_) _cachedsuper__init__formaturlkerb kerb_kwargsparsproxy ValueError_connectr_fsid) selfrrZkerberostokenuserZproxy_torZ data_proxyZ use_httpskwargs __class__]/var/www/html/Darija-Ai-API/env/lib/python3.8/site-packages/fsspec/implementations/webhdfs.pyr+s8.       zWebHDFS.__init__cCs|jSN)rrr#r#r$fsidxsz WebHDFS.fsidcCs0t|_|jr,ddlm}|f|j|j_dS)Nr)HTTPKerberosAuth)requestsSessionsessionrZrequests_kerberosr(rauth)rr(r#r#r$r|s  zWebHDFS._connectgetTc Ks|jt|p d}|}||j||d<td|||jj |||||d} | j dkrz$| } | dd} | dd} Wnt t fk rYn@X| d krt | n,| d krt| n| d krt| nt| | | S) Nopzsending %s with %s)methodrparamsdataallow_redirects)iiiiiZRemoteExceptionmessage exception)ZIllegalArgumentExceptionZUnsupportedOperationException)ZSecurityExceptionZAccessControlException)ZFileNotFoundException)rrcopyupdaterupperloggerdebugr+request status_codejsonrKeyErrorPermissionErrorFileNotFoundError RuntimeErrorraise_for_status) rr/r0pathr2redirectr rargsouterrmsgexpr#r#r$_calls8       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 WebHDFilerN)rrCrLrMrOrPrQr r#r#r$_opens! z WebHDFS._opencCs |d|d<|d|d<|S)Ntypelengthsize)lower)infor#r#r$ _process_infos zWebHDFS._process_infocCs t|dS)NrC)r)clsrCr#r#r$_strip_protocolszWebHDFS._strip_protocolcCs:t|}|dd|ddd|kr6|d|d<|S)NrCrusernamer)rpop)ZurlpathrFr#r#r$_get_kwargs_from_urlss   zWebHDFS._get_kwargs_from_urlscCs,|jd|d}|d}||d<||S)NZ GETFILESTATUSrC FileStatusname)rJr=rZ)rrCrFrYr#r#r$rYs z WebHDFS.infocCsv|jd|d}|dd}|D](}|||dd|d|d<q"|r`t|dd d Std d |DSdS) NZ LISTSTATUSr`Z FileStatusesra/Z pathSuffixrbcSs|dS)Nrbr#)ir#r#r$zWebHDFS.ls..)keycss|]}|dVqdS)rbNr#).0rYr#r#r$ szWebHDFS.ls..)rJr=rZrstripsorted)rrCdetailrFinfosrYr#r#r$lss z WebHDFS.lscCs|jd|d}|dS)z8Total numbers of files, directories and bytes under pathZGETCONTENTSUMMARYr`ZContentSummaryrJr=)rrCrFr#r#r$content_summaryszWebHDFS.content_summarycCsb|jd|dd}d|jkrJ||jd}|j|}||dS||dSdS)z/Checksum info of file, giving method and resultZGETFILECHECKSUMF)rCrDLocationZ FileChecksumN)rJheaders _apply_proxyr+r-rBr=)rrCrFlocationout2r#r#r$ukeys   z WebHDFS.ukeycCs|d}|dS)zGet user's home directoryZGETHOMEDIRECTORYPathro)rrFr#r#r$home_directorys zWebHDFS.home_directorycCsB|r|jd|d}n |d}|d}|dkr: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 ZGETDELEGATIONTOKEN)renewerTokenNz1No token available for this user/security contextZ urlString)rJr=r)rryrFtr#r#r$get_delegation_token s  zWebHDFS.get_delegation_tokencCs|jdd|d}|dS)z/Make token live longer. Returns new expiry timeZRENEWDELEGATIONTOKENputr0rlongro)rrrFr#r#r$renew_delegation_tokenszWebHDFS.renew_delegation_tokencCs|jdd|ddS)z Stop the token from being usefulZCANCELDELEGATIONTOKENr}r~NrJ)rrr#r#r$cancel_delegation_token szWebHDFS.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})r0rCZ permissionNr)rrCmodr#r#r$chmod$s z WebHDFS.chmodcCs>i}|dk r||d<|dk r$||d<|jdd|d|dS)zChange owning user and/or groupNownergroupSETOWNERr}r0rC)rr)rrCrrr r#r#r$chown1s 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})rCr0rPNr)rrCrPr#r#r$set_replication:s zWebHDFS.set_replicationcKs|jdd|ddS)NZMKDIRSr}rrrrCr r#r#r$mkdirHsz WebHDFS.mkdircCs(|dkr||rt|||dS)NF)existsFileExistsErrorr)rrCexist_okr#r#r$makedirsKszWebHDFS.makedirscKs|jdd||ddS)NZRENAMEr})r0rC destinationr)rZpath1Zpath2r r#r#r$mvPsz WebHDFS.mvcKs|jdd||rdndddS)NDELETEdeletetruefalse)r0rC recursiver)rrCrr r#r#r$rmSs  z WebHDFS.rmcKs||dSr%)rrr#r#r$rm_file[szWebHDFS.rm_filec Ks||}d||dtdg}z4||d}t||W5QRX|||Wn4tk rt t | |W5QRXYnXW5QRXdS)Nrcz.tmp.wb) openjoin_parentsecretsZ token_hexshutil copyfileobjr BaseExceptionrr@r)rZlpathZrpathr ZlstreamZ tmp_fnameZrstreamr#r#r$cp_file^s   zWebHDFS.cp_filecCsH|jrt|jr||}n(|jrD|jD]\}}|||d}q,|S)N)rcallableitemsreplace)rrtkvr#r#r$rsls  zWebHDFS._apply_proxy)r FNNNNNF)r-NNT)rKNTNN)F)N)NN)F)F)'__name__ __module__ __qualname____doc__strtempfile gettempdirrNrrpropertyr'rrJrT staticmethodrZ classmethodr\r_rYrnrprvrxr|rrrrrrrrrrrrs __classcell__r#r#r!r$r s\ M  # -         r csJeZdZdZfddZdddZddZd d Zd d Zd dZ Z S)rSz"A file living in HDFS over webHDFSc stj||f||}|dddkr6|dd|dddkrR|dd|dd|_|d}|dddkr|j|_tj |t t |_dS)NrQrPirNrOF) rrr6r-r^rQrCtargetosrruuiduuid4)rfsrCr rNr!r#r$rys   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)r2rrT)rr+postrtbuffergetvaluerB)rfinalrFr#r#r$ _upload_chunks zWebHDFile._upload_chunkcCs|j}d|jkrd\}}nd\}}d|d<|jj|||jfddi|}|j|jd}d |jkr|jjj |d d id }| |jjd d|jfddi|}|j|jd|_ dS)zCreate remote file/uploada)APPENDPOST)ZCREATEPUTr overwriterDFrqwrr)rrrrN) r r6rLrrJrCrsrrr+r}rBrt)rr r/r0rFrtrur#r#r$_initiate_uploads    zWebHDFile._initiate_uploadcCst|d}t|j|}||ks(||jkr,dS|jjd|j|||dd}|d|jkr|jd}|jj |j |}|j S|j SdS)NrrfZOPENF)rCoffsetrVrDrq) maxminrWrrJrCrBrrr+r-rscontent)rstartendrFrtrur#r#r$ _fetch_ranges"    zWebHDFile._fetch_rangecCs|j|j|jdSr%)rrrCrr&r#r#r$commitszWebHDFile.commitcCs|j|jdSr%)rrrCr&r#r#r$discardszWebHDFile.discard)F) rrrrrrrrrrrr#r#r!r$rSvs rS)loggingrrrrr contextlibr urllib.parserr)specrrutilsrr getLoggerr9r rSr#r#r#r$s   d