o hA@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 webHDFSFNTc  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 token delegationzQIf passing a password, the user must also beset in order to set up the basic-authz user.namedoaszJIf using Kerberos auth, do not specify the user, this is handled by kinit.webhdfs_)_cachedsuper__init__urlkerb kerb_kwargsparsproxy ValueErroruserpassword session_certsession_verify_connectr_fsid)selfhostportkerberostokenrrproxy_tor data_proxy use_httpsrr kwargs __class__rr/var/www/html/construction_image-detection-poc/venv/lib/python3.10/site-packages/fsspec/implementations/webhdfs.pyr-sD9       zWebHDFS.__init__cCs|jSN)r"r#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)requestsSessionsessionrcertr verifyrrequests_kerberosr2rauthrr requests.authr3)r#r2r3rrr.r!s     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)iiiiiRemoteExceptionmessage exception)IllegalArgumentExceptionUnsupportedOperationException)SecurityExceptionAccessControlException)FileNotFoundException)_strip_protocol _apply_proxyrrcopyupdaterupperloggerdebugr6request status_codejsonrKeyErrorPermissionErrorFileNotFoundError RuntimeErrorraise_for_status) r#r?r@pathrBredirectr+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 WebHDFilerf)r#r[rdrergrhrir+rrr._opens !z WebHDFS._opencCs |d|d<|d|d<|S)Ntypelengthsize)lower)inforrr. _process_infos zWebHDFS._process_infocCs t|dS)Nr[)r)clsr[rrr.rLs zWebHDFS._strip_protocolcCs:t|}|dd|ddd|vr|d|d<|S)Nr[protocolusernamer)rpop)urlpathr^rrr._get_kwargs_from_urlss   zWebHDFS._get_kwargs_from_urlscCs,|jd|d}|d}||d<||S)N GETFILESTATUSr[ FileStatusname)rbrUrr)r#r[r^rqrrr.rq s  z WebHDFS.infocCsr|jd|d}|dd}|D]}|||dd|d|d<q|r0t|dd d Std d |DS) N LISTSTATUSrz FileStatusesr{/ pathSuffixr|cSs|dS)Nr|r)irrr.szWebHDFS.ls..)keycss|]}|dVqdS)r|Nr).0rqrrr. szWebHDFS.ls..)rbrUrrrstripsorted)r#r[detailr^infosrqrrr.lss z WebHDFS.lscCs|jd|d}|dS)z8Total numbers of files, directories and bytes under pathGETCONTENTSUMMARYrzContentSummaryrbrU)r#r[r^rrr.content_summarys zWebHDFS.content_summarycCs^|jd|dd}d|jvr%||jd}|j|}||dS||dS)z/Checksum info of file, giving method and resultGETFILECHECKSUMF)r[r\Location FileChecksum)rbheadersrMr6r<rZrU)r#r[r^locationout2rrr.ukeys    z WebHDFS.ukeycCs|d}|dS)zGet user's home directoryGETHOMEDIRECTORYPathr)r#r^rrr.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 GETDELEGATIONTOKEN)renewerTokenNz1No token available for this user/security context urlString)rbrUr)r#rr^trrr.get_delegation_token0s  zWebHDFS.get_delegation_tokencCs|jdd|d}|dS)z/Make token live longer. Returns new expiry timeRENEWDELEGATIONTOKENputr@r'longr)r#r'r^rrr.renew_delegation_tokenAs zWebHDFS.renew_delegation_tokencCs|jdd|ddS)z Stop the token from being usefulCANCELDELEGATIONTOKENrrNrb)r#r'rrr.cancel_delegation_tokenFszWebHDFS.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 SETPERMISSIONr)r@r[ permissionNr)r#r[modrrr.chmodJs z WebHDFS.chmodcCs>i}|dur ||d<|dur||d<|jdd|d|dS)zChange owning user and/or groupNownergroupSETOWNERrr@r[)rr)r#r[rrr+rrr.chownWs 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. SETREPLICATIONr)r[r@rhNr)r#r[rhrrr.set_replication`s zWebHDFS.set_replicationcKs|jdd|ddS)NMKDIRSrrrr#r[r+rrr.mkdirnsz WebHDFS.mkdircCs(|dur ||r t|||dS)NF)existsFileExistsErrorr)r#r[exist_okrrr.makedirsqszWebHDFS.makedirscKs|jdd||ddS)NRENAMEr)r@r[ destinationr)r#path1path2r+rrr.mvvz WebHDFS.mvcKs|jdd||r dndddS)NDELETEdeletetruefalse)r@r[ recursiver)r#r[rr+rrr.rmys   z WebHDFS.rmcKs||dSr/)rrrrr.rm_fileszWebHDFS.rm_filec Ks||_}d||dtdg}z$||d}t||Wdn1s.wY|||Wn!ty[t t  | |Wd1sUwYwWddS1sgwYdS)Nrz.tmp.wb) openjoin_parentsecrets token_hexshutil copyfileobjr BaseExceptionrrXr)r#lpathrpathr+lstream tmp_fnamerstreamrrr.cp_files"      "zWebHDFS.cp_filecCsJ|jrt|jr||}|S|jr#|jD] \}}|||d}q|S)N)rcallableitemsreplace)r#rkvrrr.rMs zWebHDFS._apply_proxy) r FNNNNNNFNT)r<NNT)rcNTNNFr/)NN)'__name__ __module__ __qualname____doc__strtempfile gettempdirrfrtrpropertyr1r!rbrl staticmethodrr classmethodrLrxrqrrrrrrrrrrrrrrrrrM __classcell__rrr,r.r sd e  $ -          r csJeZdZdZfddZdddZddZd d Zd d Zd dZ Z S)rkz"A file living in HDFS over webHDFSc stj||fi||}|dddur|dd|dddur+|dd|dd|_|d}|dddurQ|j|_tj |t t |_dSdS)NrirhirfrgF) rrrNr<rvrir[targetosrruuiduuid4)r#fsr[r+rfr,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)rBrT)rr6postrbuffergetvaluerZ)r#finalr^rrr. _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)CREATEPUTr overwriter\Frwrr)rrrN) r+rNrdrrbr[rMrr6rrZr)r#r+r?r@r^rrrrr._initiate_uploads    zWebHDFile._initiate_uploadcCst|d}t|j|}||ks||jkrdS|jjd|j|||dd}|d|jvr@|jd}|jj |j |}|j S|j S)NrOPENF)r[offsetrnr\r) maxminrorrbr[rZrr6r<rMcontent)r#startendr^rrrrr. _fetch_ranges    zWebHDFile._fetch_rangecCs|j|j|jdSr/)rrr[rr0rrr.commitrzWebHDFile.commitcCs|j|jdSr/)rrr[r0rrr.discardszWebHDFile.discardr) rrrrrrrrrrrrrr,r.rks  rk)loggingrrrrr contextlibr urllib.parserr4specrrutilsrr getLoggerrQr rkrrrr.s"