o ˳i"F @sUdZddlZddlZddlZddlZddlZddlZddlmZddl m Z ddl m Z ddl m Z mZddlmZmZddlZddlmZdd lmZdd lmZdd lmZmZdd lmZdd lmZddlm Z m!Z!m"Z"m#Z#m$Z$m%Z%m&Z&m'Z'm(Z(m)Z)m*Z*m+Z+m,Z,m-Z-m.Z.m/Z/m0Z0m1Z1m2Z2m3Z3m4Z4m5Z5m6Z6m7Z7m8Z8m9Z9m:Z:m;Z;mZ>m?Z?m@Z@mAZAmBZBmCZCmDZDmEZEmFZFmGZGmHZHmIZImJZJmKZKmLZLmMZMmNZNmOZOddlPmQZQddlRmSZSddlTmUZUmVZVmWZWeXeYZZe rddl[m\Z\Gdddej]Z^Gddde^Z_de_iZ`eaebece^fedd<deQdefddZedeQdefddZfdeQdefdd Zgd0deQd!ebdBdefd"d#ZhdeQdefd$d%Zieeeeefefefehehegegeid& Zjeaebed'effedd(<ekhd)ZlGd*d+d+eLZmGd,d-d-eNZnGd.d/d/eKZodS)1zbFileIO implementation for reading and writing table files that uses fsspec compatible filesystems.N)Callable)copy) lru_cache) TYPE_CHECKINGAny) ParseResulturlparse)AbstractFileSystemLocalFileSystem) HTTPError)TOKENURI) AUTH_MANAGER) SignError)0ADLS_ACCOUNT_HOSTADLS_ACCOUNT_KEYADLS_ACCOUNT_NAME ADLS_ANONADLS_CLIENT_IDADLS_CLIENT_SECRETADLS_CONNECTION_STRINGADLS_CREDENTIALADLS_SAS_TOKENADLS_TENANT_ID ADLS_TOKENAWS_ACCESS_KEY_IDAWS_PROFILE_NAME AWS_REGIONAWS_SECRET_ACCESS_KEYAWS_SESSION_TOKEN GCS_ACCESSGCS_CACHE_TIMEOUTGCS_CONSISTENCYGCS_DEFAULT_LOCATIONGCS_PROJECT_IDGCS_REQUESTER_PAYSGCS_SERVICE_HOSTGCS_SESSION_KWARGS GCS_TOKENGCS_VERSION_AWARE HF_ENDPOINTHF_TOKENS3_ACCESS_KEY_ID S3_ANONYMOUSS3_CONNECT_TIMEOUT S3_ENDPOINTS3_FORCE_VIRTUAL_ADDRESSINGS3_PROFILE_NAME S3_PROXY_URI S3_REGIONS3_REQUEST_TIMEOUTS3_SECRET_ACCESS_KEYS3_SESSION_TOKEN S3_SIGNERS3_SIGNER_ENDPOINTS3_SIGNER_ENDPOINT_DEFAULT S3_SIGNER_URIFileIO InputFile InputStream OutputFile OutputStream) Properties) strtobool)get_first_property_valueget_header_propertiesproperty_as_bool) AWSRequestc@sHeZdZUdZeed<deddfddZejddd e ddfd d Z dS) S3RequestSignerz+Abstract base class for S3 request signers. propertiesreturnNcC ||_dSNrHselfrHrOG/home/ubuntu/.local/lib/python3.10/site-packages/pyiceberg/io/fsspec.py__init__k zS3RequestSigner.__init__requestrF_cKsdSrKrO)rNrSrTrOrOrP__call__nszS3RequestSigner.__call__) __name__ __module__ __qualname____doc__rA__annotations__rQabcabstractmethodrrUrOrOrOrPrGfs rGcsLeZdZUdZejed<deddffdd Zdd d e ddfd d Z Z S) S3V4RestSignerzQAn S3 request signer that uses an external REST signing service to sign requests._sessionrHrINcst|t|_dSrK)superrQrequestsSessionr^rM __class__rOrPrQxs zS3V4RestSigner.__init__rSrFrTc KsF|jt|jtd}|jtt}i}d}|jt}r%|}n |jt }r2d|}|r8||d<| t |j|j |j d|jdd|jDd} |jj|d||| d} z | | } Wnty} z td | jd | | d} ~ ww| d D]\} }|j| d |q| d |_dS)N/zBearer Authorization client_regioncSsi|]\}}||gqSrOrO).0keyvalrOrOrP sz+S3V4RestSigner.__call__..)methodregionuriheaders)rnjsonzFailed to sign request z: rnz, rm)rHgetr;rrstripr9r:r auth_headerr updaterDrkcontexturlrnitemsr^poststripraise_for_statusror r status_code add_headerjoin)rNrSrT signer_urlsigner_endpointsigner_headersrr auth_managertoken signer_bodyresponse response_jsonerhvaluerOrOrPrU|s6    zS3V4RestSigner.__call__) rVrWrXrYr`rarZrArQrrU __classcell__rOrOrbrPr]ss  r]SIGNERSrTrIcCs tddS)NT) auto_mkdirr )rTrOrOrP_filerRrrHcCsddlm}|tt|ttt|ttt|t t t|t t d}i}i}|t }rStd|t|}rL||}||d<ddlm}||d<ntd||t}ra||d |d <|t} rnt| |d <|t} r{t| |d <|t} rd di|d<|t} rt| } nd} | ||d}t|tt}r||d<|di|}|D]\}}|jjj j!|dd|jjj j"||ddq|S)Nr) S3FileSystem) endpoint_urlaws_access_key_idaws_secret_access_keyaws_session_token region_namezLoading signer %szbefore-sign.s3)UNSIGNEDsignature_versionzSigner not available: )httphttpsproxiesconnect_timeout read_timeoutaddressing_stylevirtuals3F)anon client_kwargs config_kwargsprofilei) unique_idrO)#s3fsrrpr0rCr-rr6rr7r r4rr8loggerinforbotocorer ValueErrorr3r/floatr5r1r.rBr2rrvrmetaevents unregister register_last)rHrrrregister_eventssigner signer_clsr proxy_urirrequest_timeout_force_virtual_addressing s3_anonymousr s3_fs_kwargs profile_namefs event_nameevent_functionrOrOrP_s3sN            rcCstddlm}||t|td|t|td|tt|t dt |t d|t |tt|tdd S)Nr) GCSFileSystem full_controlnoneFz{}) projectaccessr consistency cache_timeoutrequester_payssession_kwargsrdefault_location version_aware)gcsfsrrpr%r!r)r#r"rEr&roloadsr(r'r$r*)rHrrOrOrP_gss     rhostnamec sddlm}ddlmddlm}dd|DD]\}}t|vr.|dd|t<t |vr6||t <q|rFt|vrF|dd|t<Gfdd d |}| t }r\||}n| t }|| t || t| t| t | t| t| t| t| td S) Nr)AzureBlobFileSystem AccessToken)AsyncTokenCredentialcSs4i|]\}}|tdr|tdd|qS).) startswithrreplace)rgrhrrOrOrPrjs z_adls..rcs<eZdZdZdeddfddZdededffd d ZdS) z$_adls..StaticTokenCredentiali token_stringrINcSrJrK)_token)rNrrOrOrPrQ rRz-_adls..StaticTokenCredential.__init__scopeskwargscs(ddl}t||j}|j|S)Nr)timeint_DEFAULT_EXPIRY_SECONDSr)rNrrr expires_onrrOrP get_tokens z._adls..StaticTokenCredential.get_token)rVrWrXrstrrQrrrOrrOrPStaticTokenCredential sr) connection_string credential account_name account_key sas_token tenant_id client_id client_secret account_hostr)adlfsrazure.core.credentialsrazure.core.credentials_asyncrrvrsplitrrprrrrrrrrr) rHrrrrhrrrrrOrrP_adlss>       rcCs$ddlm}||t|tdS)Nr) HfFileSystem)endpointr)huggingface_hubrrpr+r,)rHrrOrOrP_hf*s r) rfilers3as3nabfsabfssgsgcshf. SCHEME_TO_FS>rwasbrwasbscsZeZdZdZdedeffdd ZdefddZde fd d Z dd e de fd dZ Z S)FsspecInputFilezAn input file implementation for the FsspecFileIO. Args: location (str): A URI to a file location. fs (AbstractFileSystem): An fsspec filesystem instance. locationrc||_tj|ddSN)r_fsr_rQrNrrrbrOrPrQKzFsspecInputFile.__init__rIcCB|j|j}|d}r|S|d}r|Std|jz.Return the total length of the file, in bytes.SizesizezCannot retrieve object info: rrrrp RuntimeErrorrN object_inforrOrOrP__len__O zFsspecInputFile.__len__cC|j|jSz"Check whether the location exists.rlexistsrrNrOrOrPexistsXzFsspecInputFile.existsTseekablec CsRz |j|jdWSty(}z|jr||ttjttj|j|d}~ww)aHCreate an input stream for reading the contents of the file. Args: seekable: If the stream should support seek, or if it is consumed sequential. Returns: OpenFile: An fsspec compliant file-like object. Raises: FileNotFoundError: If the file does not exist. rbN) ropenrFileNotFoundErrorfilenameerrnoENOENTosstrerror)rNrrrOrOrPr\s  &zFsspecInputFile.open)T)rVrWrXrYrr rQrrboolr r>rrrOrOrbrPrCs  rcsheZdZdZdedeffdd ZdefddZde fd d Z dd e de fd dZ de fddZZS)FsspecOutputFilezAn output file implementation for the FsspecFileIO. Args: location (str): A URI to a file location. fs (AbstractFileSystem): An fsspec filesystem instance. rrcrrrrrbrOrPrQwrzFsspecOutputFile.__init__rIcCrrrrrOrOrPr{rzFsspecOutputFile.__len__cCrr r r rOrOrPr rzFsspecOutputFile.existsF overwritecCs,|s|rtd|j|j|jdS)a/Create an output stream for reading the contents of the file. Args: overwrite (bool): Whether to overwrite the file if it already exists. Returns: OpenFile: An fsspec compliant file-like object. Raises: FileExistsError: If the file already exists at the location and overwrite is set to False. Note: If overwrite is set to False, a check is first performed to verify that the file does not exist. This is not thread-safe and a possibility does exist that the file can be created by a concurrent process after the existence check yet before the output stream is created. In such a case, the default behavior will truncate the contents of the existing file when opening the output stream. z)Cannot create file, file already exists: wb)r FileExistsErrorrrr)rNrrOrOrPcreates zFsspecOutputFile.createcCst|j|jdS)zAReturn a new FsspecInputFile for the location at `self.location`.rr)rrrr rOrOrP to_input_fileszFsspecOutputFile.to_input_file)F)rVrWrXrYrr rQrrrr r@rrrrrOrOrbrPros rcseZdZdZdeffdd ZdedefddZdede fd d Z dee Be Bdd fd d Z dddefddZddeded BdefddZddeded BdefddZdeeeffddZdeeefdd fddZZS) FsspecFileIOz)A FileIO implementation that uses fsspec.rHcs&tt|_t|_tj|ddS)NrL)dictr _scheme_to_fs threadinglocal_thread_localsr_rQrMrbrOrPrQs  zFsspecFileIO.__init__rrIcCt|}||}t||dS)a Get an FsspecInputFile instance to read bytes from the file at the given location. Args: location (str): A URI or a path to a local file. Returns: FsspecInputFile: An FsspecInputFile instance for the given location. r)r_get_fs_from_urirrNrrmrrOrOrP new_input  zFsspecFileIO.new_inputcCr&)a Get an FsspecOutputFile instance to write bytes to the file at the given location. Args: location (str): A URI or a path to a local file. Returns: FsspecOutputFile: An FsspecOutputFile instance for the given location. r)rr'rr(rOrOrP new_outputr*zFsspecFileIO.new_outputNcCs:t|ttfr |j}n|}t|}||}||dS)a9Delete the file at the given location. Args: location (Union[str, InputFile, OutputFile]): The URI to the file--if an InputFile instance or an OutputFile instance is provided, the location attribute for that instance is used as the location to delete. N) isinstancer=r?rrr'rm)rNr str_locationrmrrOrOrPdeletes  zFsspecFileIO.deletermrcCs&|jtvr ||j|jS||jS)zOGet a filesystem from a parsed URI, using hostname for ADLS account resolution.)scheme _ADLS_SCHEMESget_fsr)rNrmrOrOrPr's  zFsspecFileIO._get_fs_from_urir0rcCs(t|jds t|j|j_|j||S)z:Get a filesystem for a specific scheme, cached per thread. get_fs_cached)hasattrr%r_get_fsr3rNr0rrOrOrPr2s zFsspecFileIO.get_fscCs<||jvr td||tvrt|j|S|j||jS)z'Get a filesystem for a specific scheme.z%No registered filesystem for scheme: )r"rr1rrHr6rOrOrPr5s  zFsspecFileIO._get_fscCst|j}|d=|S)zBCreate a dictionary of the FsSpecFileIO fields used when pickling.r%)r__dict__)rN fileio_copyrOrOrP __getstate__s zFsspecFileIO.__getstate__statecCs||_t|_dS)z3Deserialize the state into a FsSpecFileIO instance.N)r7r#r$r%)rNr:rOrOrP __setstate__szFsspecFileIO.__setstate__rK)rVrWrXrYrArQrrr)rr+r=r?r/r r'r2r5r!rr9r;rrOrOrbrPr s   "r rK)prYr[rrologgingrr#collections.abcrr functoolsrtypingrr urllib.parserrr`fsspecr fsspec.implementations.localr r pyiceberg.catalogr rpyiceberg.catalog.rest.authrpyiceberg.exceptionsr pyiceberg.iorrrrrrrrrrrrrrrr r!r"r#r$r%r&r'r(r)r*r+r,r-r.r/r0r1r2r3r4r5r6r7r8r9r:r;r<r=r>r?r@pyiceberg.typedefrApyiceberg.typesrBpyiceberg.utils.propertiesrCrDrE getLoggerrVrbotocore.awsrequestrFABCrGr]rr!rtyperZrrrrrr frozensetr1rrr rOrOrOrPsb         2    .=3  ,4