o iB @sUdZddlZddlZddlZddlZddlZddlZddlmZddl m Z ddl m Z ddl m Z mZddlmZddlZddlmZdd lmZdd lmZdd lmZmZdd lmZdd lmZddlmZm 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@mAZAmBZBmCZCmDZDmEZEmFZFmGZGmHZHmIZImJZJmKZKmLZLmMZMmNZNddlOmPZPddlQmRZRddlSmTZTmUZUmVZVeWeXZYe rddlZm[Z[Gdddej\Z]Gddde]Z^de^iZ_e`eaebe]fecd<dePdefddZddePdefddZedePdefdd ZfdePdefd!d"ZgdePdefd#d$Zhededeeeeeeegegefefehd% ZiGd&d'd'eKZjGd(d)d)eMZkGd*d+d+eJZldS),zbFileIO implementation for reading and writing table files that uses fsspec compatible filesystems.N)Callable)copy) lru_cache) TYPE_CHECKINGAny)urlparse)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 ||_dSNrGselfrGrNQ/home/ubuntu/transcripts/venv/lib/python3.10/site-packages/pyiceberg/io/fsspec.py__init__k zS3RequestSigner.__init__requestrE_cKsdSrJrN)rMrRrSrNrNrO__call__nszS3RequestSigner.__call__) __name__ __module__ __qualname____doc__r@__annotations__rPabcabstractmethodrrTrNrNrNrOrFfs rFcsLeZdZUdZejed<deddffdd Zdd d e ddfd d Z Z S) S3V4RestSignerzQAn S3 request signer that uses an external REST signing service to sign requests._sessionrGrHNcst|t|_dSrJ)superrPrequestsSessionr]rL __class__rNrOrPxs zS3V4RestSigner.__init__rRrErSc 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|]\}}||gqSrNrN).0keyvalrNrNrO sz+S3V4RestSigner.__call__..)methodregionuriheaders)rmjsonzFailed to sign request z: rmz, rl)rGgetr:r rstripr8r9r auth_headerr updaterCrjcontexturlrmitemsr]poststripraise_for_statusrnr r status_code add_headerjoin)rMrRrS signer_urlsigner_endpointsigner_headersrq auth_managertoken signer_bodyresponse response_jsonergvaluerNrNrOrT|s6    zS3V4RestSigner.__call__) rUrVrWrXr_r`rYr@rPrrT __classcell__rNrNrarOr\ss  r\SIGNERSrSrHcCs tddS)NT) auto_mkdirr )rSrNrNrO_filerQrrGcCsddlm}|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_idrN)#s3fsrror/rBr,rr5rr6rr3rr7loggerinforbotocorer ValueErrorr2r.floatr4r0r-rAr1rrurmetaevents unregister register_last)rGrrrregister_eventssigner signer_clsr proxy_urirrequest_timeout_force_virtual_addressing s3_anonymousr s3_fs_kwargs profile_namefs event_nameevent_functionrNrNrO_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)gcsfsrror$r r(r"r!rDr%rnloadsr'r&r#r))rGrrNrNrO_gss     rc sddlm}ddlmddlm}dd|DD]\}}t|vr.|dd|t<t |vr6||t <qGfdd d |}| t }rM||}n| t }|| t || t| t| t | t| t| t| t| td S) Nr)AzureBlobFileSystem AccessToken)AsyncTokenCredentialcSs4i|]\}}|tdr|tdd|qS).) startswithrreplace)rfrgrrNrNrOris z_adls..rcs<eZdZdZdeddfddZdededffd d ZdS) z$_adls..StaticTokenCredentiali token_stringrHNcSrIrJ)_token)rMrrNrNrOrP rQz-_adls..StaticTokenCredential.__init__scopeskwargscs(ddl}t||j}|j|S)Nr)timeint_DEFAULT_EXPIRY_SECONDSr)rMrrr expires_onrrNrO get_token s z._adls..StaticTokenCredential.get_token)rUrVrWrstrrPrrrNrrNrOStaticTokenCredentialsr) connection_string credential account_name account_key sas_token tenant_id client_id client_secret account_hostr)adlfsrazure.core.credentialsrazure.core.credentials_asyncrrursplitrrorrrrrrrrr)rGrrrgrrrrrNrrO_adlss:      rcCs$ddlm}||t|tdS)Nr) HfFileSystem)endpointr)huggingface_hubrror*r+)rGrrNrNrO_hf&s r) rfilers3as3nabfsabfssgsgcshfcsZeZdZdZdedeffdd 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^rPrMrrrarNrOrPEzFsspecInputFile.__init__rHcCB|j|j}|d}r|S|d}r|Std|jz.Return the total length of the file, in bytes.SizesizezCannot retrieve object info: rrrro RuntimeErrorrM object_inforrNrNrO__len__I zFsspecInputFile.__len__cC|j|jSz"Check whether the location exists.rlexistsrrMrNrNrOexistsRzFsspecInputFile.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)rMr rrNrNrOr Vs  &zFsspecInputFile.open)T)rUrVrWrXrrrPrrboolrr=r rrNrNrarOr=s  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. rrcrrrrrarNrOrPqrzFsspecOutputFile.__init__rHcCrrrrrNrNrOrurzFsspecOutputFile.__len__cCrrrrrNrNrOr~r zFsspecOutputFile.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)rFileExistsErrorrrr )rMrrNrNrOcreates zFsspecOutputFile.createcCst|j|jdS)zAReturn a new FsspecInputFile for the location at `self.location`.rr)rrrrrNrNrO to_input_fileszFsspecOutputFile.to_input_file)F)rUrVrWrXrrrPrrrrr?rrrrrNrNrarOris rcseZdZdZdeffdd ZdedefddZdede fd d Z dee Be Bdd fd d Z dedefddZdedefddZdeeeffddZdeeefdd fddZZS) FsspecFileIOz)A FileIO implementation that uses fsspec.rGcs.i|_|jtt|_tj|ddS)NrK) _scheme_to_fsrr SCHEME_TO_FS threadinglocal_thread_localsr^rPrLrarNrOrPs  zFsspecFileIO.__init__rrHcC t|}||j}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)rget_fsschemerrMrrlrrNrNrO 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"r#rr$rNrNrO new_outputr&zFsspecFileIO.new_outputNcCs<t|ttfr |j}n|}t|}||j}||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"r#rm)rMr str_locationrlrrNrNrOdeletes  zFsspecFileIO.deleter#cCs&t|jds t|j|j_|j|S)z:Get a filesystem for a specific scheme, cached per thread. get_fs_cached)hasattrr r_get_fsr,rMr#rNrNrOr"s  zFsspecFileIO.get_fscCs(||jvr td||j||jS)z'Get a filesystem for a specific scheme.z%No registered filesystem for scheme: )rrrGr/rNrNrOr.s zFsspecFileIO._get_fscCst|j}|d=|S)zBCreate a dictionary of the FsSpecFileIO fields used when pickling.r )r__dict__)rM fileio_copyrNrNrO __getstate__s zFsspecFileIO.__getstate__statecCs||_t|_dS)z3Deserialize the state into a FsSpecFileIO instance.N)r0rrr )rMr3rNrNrO __setstate__szFsspecFileIO.__setstate__)rUrVrWrXr@rPrrr%rr'r<r>r+rr"r.dictrr2r4rrNrNrarOrs  "r)mrXrZrrnloggingrrcollections.abcrr functoolsrtypingrr urllib.parserr_fsspecrfsspec.implementations.localr r pyiceberg.catalogr r pyiceberg.catalog.rest.authrpyiceberg.exceptionsr pyiceberg.iorrrrrrrrrrrrrrrrr r!r"r#r$r%r&r'r(r)r*r+r,r-r.r/r0r1r2r3r4r5r6r7r8r9r:r;r<r=r>r?pyiceberg.typedefr@pyiceberg.typesrApyiceberg.utils.propertiesrBrCrD getLoggerrUrbotocore.awsrequestrEABCrFr\rr5rtyperYrrrrrrrrrrNrNrNrOs`          2    .=/ ,4