o i @sdZddlmZddlZddlZddlZddlZddlZddlZddl Z ddl m Z ddl m Z mZmZzddlZddlZddlZddlZWn eyQdZYnwddlZddlZddlZddlmZerrddlmZdd lm Z e!e"Z#d Z$ d Z% d Z& d Z'dZ(dZ)dZ*dZ+dZ,GdddZ-de.vre-Z/GdddZ0ddZ1ddZ2ddZ3de*e%ddddddf dd Z4d!d"Z5d#d$Z6Gd%d&d&e7Z8d'd(Z9Gd)d*d*ej:Z;Gd+d,d,ej:Z 1   2 3 4 d@d5d6Z?d1d7d8d9d:d;Z@dd?ZBdS)AzDImplements file-like objects for reading and writing from/to AWS S3.) annotationsN)inf)CallableList TYPE_CHECKINGT) constants)S3Client)BufferiPi l)s3s3ns3us3aizs3.amazonaws.comi)zs3://my_bucket/my_keyz&s3://my_key:my_secret@my_bucket/my_keyz8s3://my_key:my_secret@my_server:my_port@my_bucket/my_key InvalidRangec@seZdZddZdddZdS) RetrycCs$d|_d|_tjjg|_dg|_dS)N NoSuchUpload)attempts sleep_secondsbotocore exceptionsEndpointConnectionErrorclient_error_codesselfrA/home/ubuntu/.local/lib/python3.10/site-packages/smart_open/s3.py__init__Us  zRetry.__init__fnrc Cst|jD]l}z|WSt|jy9}ztd||j|dt|t|j WYd}~qd}~wt jj yq}z+|j d d}||jvrPtd||j|dt|t|j WYd}~qd}~wwtdtd|j|j)Nz+Caught non-fatal %s, retrying %d more timesErrorCodez9Caught non-fatal ClientError (%s), retrying %d more timesz0encountered too many non-fatal errors, giving upz%s failed after %d attempts)rangertuplerloggercritical exceptiontimesleeprr ClientErrorresponsegetrIOErrorfunc)rrattempterr error_coderrr_do[s6       z Retry._doN)rr)__name__ __module__ __qualname__rr1rrrrrTsr MISSING_DEPSc@s eZdZdZddZddZdS)_ClientWrapperaxWraps a client to inject the appropriate keyword args into each method call. The keyword args are a dictionary keyed by the fully qualified method name. For example, S3.Client.create_multipart_upload. See https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/s3.html#client This wrapper behaves identically to the client otherwise. cCs||_||_dSN)clientkwargs)rr8r9rrrrs z_ClientWrapper.__init__cCs0t|j|}|jd|i}tj|fi|S)Nz S3.Client.%s)getattrr8r9r+ functoolspartial)r method_namemethodr9rrr __getattr__s z_ClientWrapper.__getattr__N)r2r3r4__doc__rr?rrrrr6s r6c Cstj|}|jtvs Jt}t}d}d\}}|j|j}d|vr@| dd\}} d|vr@| dd\} } d| vr@| | }}| }| dd\} } d| vrfd| vrfd}| d\}}| dd\}}t |}nd| vrtd}| d\}}n| }t |j|| |||||dS) NFNN@r:/T)scheme bucket_idkey_idporthostordinary_calling_format access_id access_secret) smart_openutils safe_urlsplitrESCHEMES DEFAULT_PORT DEFAULT_HOSTnetlocpathsplitintdict) uri_as_string split_urirHrIrJrKrLuri maybe_authrestmaybe_id maybe_secretheadrG host_portrFrrr parse_urisD     racstfdd}d}|dur(|ds|dr(td|jdddn|drA|drA||d|dd |jddd|durZ|d tkrZtd |jdd |fS|d tkrz|d dkrid}nd}||d|d|jdd |fS)a`Consolidates the parsed Uri with the additional parameters. This is necessary because the user can pass some of the parameters can in two different ways: 1) Via the URI itself 2) Via the transport parameters These are not mutually exclusive, but we have to pick one over the other in a sensible way in order to proceed. cslzd}Wntyi}d<Ynwz|d}Wnty+i}|d<Ynw|jdi|dS)N client_kwargs S3.Clientr)KeyErrorupdate)r9rb init_kwargstransport_paramsrrinjects    z#_consolidate_params..injectr8NrKrLzignoring credentials parsed from URL because they conflict with transport_params["client"]. Set transport_params["client"] to None to suppress this warning.)rKrL)aws_access_key_idaws_secret_access_keyrIzignoring endpoint_url parsed from URL because they conflict with transport_params["client"]. Set transport_params["client"] to None to suppress this warning.)rIrEr httphttpsz://%(host)s:%(port)d) endpoint_url)rWr+r$warningrerR)rZrhrir8rErrgr_consolidate_paramss6     rpc svd}fdd|D}|rd}d||f}t|tt|}t|\}tjt}t|d|d|fi|S)N)multipart_upload_kwargs object_kwargsresourceresource_kwargssessionsinglepart_upload_kwargscsg|]}|vr|qSrr).0krgrr szopen_uri..zVhttps://github.com/piskvorky/smart_open/blob/develop/MIGRATING_FROM_OLDER_VERSIONS.rstzPignoring the following deprecated transport parameters: %r. See <%s> for detailsrFrG) warningswarn UserWarningrarprMrN check_kwargsopen) rZmoderh deprecateddetecteddoc_urlmessage parsed_urir9rrgropen_uris  rFc Cstdt|tjvrtd|tjf|tjkr"|dur"td|tjkr4t ||||||| | d} n"|tjkrP|rFt |||| | |d} nt |||| | d} nJd ||| _ | S) aOpen an S3 object for reading or writing. Parameters ---------- bucket_id: str The name of the bucket this object resides in. key_id: str The name of the key within the bucket. mode: str The mode for opening the object. Must be either "rb" or "wb". buffer_size: int, optional Default: 128KB The buffer size in bytes for reading. Controls memory usage. Data is streamed from a S3 network stream in buffer_size chunks. Forward seeks within the current buffer are satisfied without additional GET requests. Backward seeks always open a new GET request. For forward seek-intensive workloads, increase buffer_size to reduce GET requests at the cost of higher memory usage. min_part_size: int, optional The minimum part size for multipart uploads, in bytes. When the writebuffer contains this many bytes, smart_open will upload the bytes to S3 as a single part of a multi-part upload, freeing the buffer either partially or entirely. When you close the writer, it will assemble the parts together. The value determines the upper limit for the writebuffer. If buffer space is short (e.g. you are buffering to memory), then use a smaller value for min_part_size, or consider buffering to disk instead (see the writebuffer option). The value must be between 5MB and 5GB. If you specify a value outside of this range, smart_open will adjust it for you, because otherwise the upload _will_ fail. For writing only. Does not apply if you set multipart_upload=False. multipart_upload: bool, optional Default: `True` If set to `True`, will use multipart upload for writing to S3. If set to `False`, S3 upload will use the S3 Single-Part Upload API, which is more ideal for small file sizes. For writing only. version_id: str, optional Version of the object, used when reading object. If None, will fetch the most recent version. defer_seek: boolean, optional Default: `False` If set to `True` on a file opened for reading, GetObject will not be called until the first seek() or read(). Avoids redundant API queries when seeking before reading. range_chunk_size: int, optional Default: `None` Maximum byte range per S3 GET request when reading. When None (default), a single GET request is made for the entire file, and data is streamed from that single botocore.response.StreamingBody in buffer_size chunks. When set to a positive integer, multiple GET requests are made, each limited to at most this many bytes via HTTP Range headers. Each GET returns a new StreamingBody that is streamed in buffer_size chunks. Useful for reading small portions of large files without forcing S3-compatible systems like SeaweedFS/Ceph to load the entire file. Larger values mean fewer billable GET requests but higher load on S3 servers. Smaller values mean more GET requests but less server load per request. Values larger than the file size result in a single GET for the whole file. Affects reading only. Does not affect memory usage (controlled by buffer_size). client: object, optional The S3 client to use when working with boto3. If you don't specify this, then smart_open will create a new client for you. client_kwargs: dict, optional Additional parameters to pass to the relevant functions of the client. The keys are fully qualified method names, e.g. `S3.Client.create_multipart_upload`. The values are kwargs to pass to that method each time it is called. writebuffer: IO[bytes], optional By default, this module will buffer data in memory using io.BytesIO when writing. Pass another binary IO instance here to use it instead. For example, you may pass a file object to buffer to local disk instead of in RAM. Use this to keep RAM usage low at the expense of additional disk IO. If you pass in an open file, then you are responsible for cleaning it up after writing completes. z%rzbad mode: %r expected one of %rNz$version_id must be None when writing) version_id buffer_size defer_seekr8rbrange_chunk_size)r8rb writebuffer part_size)r8rbrFzunexpected mode: %r) r$debuglocalsr BINARY_MODESNotImplementedError WRITE_BINARY ValueError READ_BINARYReaderMultipartWriterSinglepartWritername) rFrGrrr min_part_sizemultipart_uploadrr8rbrrfileobjrrrr~9sHY     r~c Cstzt||d}|r ||d<|r||d<|jdi|WStjjy9}ztd||||f}||_||d}~ww)NBucketKey VersionIdRangez9unable to access bucket: %r key: %r version: %r error: %sr)rW get_objectrr8r)r, backend_error)r8bucketkeyversion range_stringparamserror wrapped_errorrrr_gets" rc Cs(z|jjdWSttfyYdSw)zDGiven an IOError from _get, return the 'Error' dictionary from boto.r N)rr*AttributeErrorrd)ioerrr_unwrap_ioerrors rc@s\eZdZdZ  dddZeddZddZej fd d Z dd d Z dddZ ddZ dS)_SeekableRawReaderzHRead an S3 object. This class is internal to the S3 submodule. NcCs4||_||_||_||_||_d|_d|_d|_dS)Nr)_client_bucket_key _version_id_range_chunk_size_content_length _position_body)rr8rrrrrrrrs z_SeekableRawReader.__init__cC |jduSr7)rrrrrclosed z_SeekableRawReader.closedcCs|js |jd|_dSdSr7)rrcloserrrrrs  z_SeekableRawReader.closecCs|tjvr tdtj|d}d}|tjkrtd|}n|tjkr-td||j}n |tjkr8td| }|j durV|durS|dkrS|j ddd|d}nd}n|durb||j krbd}n |dkrid}nd}|ryt |_ |j |_|jS| |||jS)zSeek to the specified position. :param int offset: The offset in bytes. :param int whence: Where the offset is from. :returns: the position after seeking. :rtype: int z"invalid whence, expected one of %rNrstartstopTF)rWHENCE_CHOICESrr WHENCE_STARTmaxWHENCE_CURRENTr WHENCE_ENDr _open_bodyioBytesIOr)roffsetwhencerr reached_eofrrrseeks8         z_SeekableRawReader.seekc Cs&|dur |dur |j}|dur(|jdur(||jd}|jdur(t||jd}tj||}zt|j|j |j |j |}WnTt y}zHt |}|dusT|dtkrUt|dd}|dure||ksq|dur}|dur}||kr}||_|_t|_n |j||ddWYd}~dSd}~wwtd||dd |dd }|tjjkrtj|d \} } } } | |_| |_|d |_dS|tjjkr d|_|d |_|d |_|durt|j|} n|dur|durtd|j|} nd} | dkr td|| t|j| |_dSdStd|)aOpen a connection to download the specified range of bytes. Store the open file handle in self._body. If no range is specified, start defaults to self._position. start and stop follow the semantics of the http range header, so a stop without a start will read bytes beginning at stop. If self._range_chunk_size is set, the S3 server is protected from open range headers and stop will be set such that at most self._range_chunk_size bytes are returned in a single GET request. As a side effect, set self._content_length. Set self._position to self._content_length if start is past end of file. Nrr!ActualObjectSizerrz%s: RetryAttempts: %dResponseMetadata RetryAttemptsHTTPStatusCode ContentRangeBody ContentLengthz2%s: discarding %d bytes to reach expected positionzUnexpected status code %r) rrrminrMrNmake_range_stringrrrrrr,rr+ _OUT_OF_RANGErVrrrrr$rrl HTTPStatusPARTIAL_CONTENTparse_content_rangeOKrlenreadr) rrrrr*rerror_responseactual_object_size status_code_ resp_startlengthexpected_positionrrrr/sr         z_SeekableRawReader._open_bodycsdkr tddkrttd d fdd }jdus0jjkrXkrX|}jt|7_||sG jdus0jjkrXks0 S) z9Read from the continuous connection with the remote peer.rzsize must be >= -1, got rreturnbytesc s|D]S}jr ztkrjWSjWSttjj t jj fyU}zt d|tkr>dn|t|WYd}~qd}~wwtdtkr_dnt|f)Nz?%s: caught %r while reading %d bytes, sleeping %ds before retryrz-%s: failed to read %d bytes after %d attempts)rrrrrtellConnectionResetErrorrr BotoCoreErrorurllib3 HTTPErrorr$rorr'r(r,r)rsecondsr/binary_collectedrsizerr retry_reads8 z+_SeekableRawReader.read..retry_readN)r)rr) rrrrrrrrwritergetvalue)rrrbinaryrrrrs$       z_SeekableRawReader.readcCd|j|jfS)Nz%smart_open.s3._SeekableReader(%r, %r)rrrrrr__str__z_SeekableRawReader.__str__rAr)r2r3r4r@rpropertyrrrrrrrrrrrrrs    6 f ArcCsx|duri}|dur*|di}d|vr"tjjdddddd |d<tjd i|}|s.Jt|||_||_||_dS) z~Created the required objects for accessing S3. Ideally, they have been already created for us and we can just reuse them.Nrcconfig@Tradaptive max_attemptsrmax_pool_connections tcp_keepaliveretriesr r ) r+rr8Configboto3r6rrr)rwr8rbrrrfrrr_initialize_boto3s     rc@seZdZdZdeejddddfddZddZdd Z d)d d Z d)d dZ ddZ d)ddZ ddZejfddZddZd*ddZddZddZdd Zd)d!d"Zd)d#d$Zd%d&Zd'd(ZdS)+rz]Reads bytes from S3. Implements the io.BufferedIOBase interface of the standard library.NFc Csp||_||_t|||||t|j|||j| d|_d|_tj ||_ d|_ ||_ d|_ |s6|ddSdS)N)rrF)r _buffer_sizerrr _raw_reader _current_posrM bytebuffer ByteBuffer_buffer_eof_line_terminator_seek_initializedr) rrrrrline_terminatorrr8rbrrrrrs$ zReader.__init__cCstddS)zFlush and close this stream. close: calledN)r$rrrrrrs z Reader.closecCdS)z+Return True if the stream can be read from.Trrrrrreadable#zReader.readablercCsn|dkrdS|dkr||j}|jj|_|St|j|kr&||S|jr-|S||||S)z6Read up to size bytes from the object and return them.r) _read_from_bufferr rrr rr r _fill_buffer)rroutrrrr's    z Reader.readcCs |j|dS)zThis is the same as read().)r)rrrrrrread1@ z Reader.read1cCs.|t|}|s dS||dt|<t|S)zLRead up to len(b) bytes into b, and return the number of bytes read.rN)rr)rbdatarrrreadintoDs zReader.readintocCs|dkrtdt}|jrt|jdksD|j|j}|||j t|7_ | |jr6 | S| |jrt|jdkr| S)zCRead up to and including the next newline. Returns the bytes read.rz(limits other than -1 not implemented yetr) rrrrrr readlinerrr endswithrr)rlimitline line_partrrrr Ms   zReader.readlinecCr)zyIf False, seek(), tell() and truncate() will raise IOError. We offer only seek support, and no truncate support.TrrrrrseekablebzReader.seekablecCs|tjkr tj}||j7}|tjkr2||jkr2|jt|j}||kr2|j||j||_|jS|jr?|tjkr?||jksL|j |||_|j |j|jj k|_ d|_|jS)zSeek to the specified position. :param int offset: The offset in bytes. :param int whence: Where the offset is from. Returns the position after seeking.T) rrrr rr rrr remptyrr)rrr buffer_endrrrrhs   z Reader.seekcC|jS)z,Return the current position within the file.)r rrrrrz Reader.tellcCtjz Unsupported.rUnsupportedOperationrrrrtruncater*zReader.truncatecCr+r,r-rrrrdetachr*z Reader.detachcCr)z Do nothing.Nrrrrr terminaterzReader.terminatecCs6|sJd||j|j}|jdur||jS|S)Create an **independent** `boto3.s3.Object` instance that points to the same S3 object as this instance. Changes to the returned object will not affect the current instance. *resource must be a boto3.resource instanceN)ObjectrrrVersion)rrsobjrrrto_boto3s   zReader.to_boto3cCs8|dkr|nt|j}|j|}|jt|7_|S)z:Remove at most size bytes from our buffer and return them.r)rr rr )rrpartrrrrs zReader._read_from_buffercCsnt||jj}t|j|kr1|js3|j|j}|dkr%td|d|_t|j|kr5|jrdSdSdSdS)Nrz$%s: reached EOF while filling bufferT) rr  _chunk_sizerrfillr r$r)rr bytes_readrrrrs $zReader._fill_buffercCr)Nzsmart_open.s3.Reader(%r, %r)rrrrrrrzReader.__str__cCsd|j|j|j|j|jfS)NzZsmart_open.s3.Reader(bucket=%r, key=%r, version_id=%r, buffer_size=%r, line_terminator=%r))rrrrrrrrr__repr__szReader.__repr__rr7)r2r3r4r@DEFAULT_BUFFER_SIZErBINARY_NEWLINErrrrrrr r%rrrr/r0r1r7rrrr<rrrrrs6 %   !    rc@seZdZdZdZedddfd.ddZddZd d Ze d d Z d dZ ddZ e jfddZd/ddZddZddZd0ddZdd Zd!d"Zd1d$d%Zd&d'Zd(d)Zd*d+Zd,d-ZdS)2rzuWrites bytes to S3 using the multi part API. Implements the io.BufferedIOBase interface of the standard library.Nrio.BytesIO | Nonec Cstj|tt}||krtd|d||}||_t||||||||zt j |j j ||d}t |d|_WntjjyU} z td|| f| d} ~ ww|dur`t|_n||_d|_d|_g|_dS)Nzadjusting part_size from z to rUploadIdz=the bucket %r does not exist, or is forbidden for access (%r)r)rMrNclamp MIN_PART_SIZE MAX_PART_SIZEr$ro _part_sizerr;r<rcreate_multipart_uploadRETRYr1 _upload_idrr8r)rrr_buf _total_bytes _total_parts_parts) rrrrr8rbr adjusted_psr<rrrrrs@   zMultipartWriter.__init__cCdSr7rrrrrflushzMultipartWriter.flushcCstd|jr dS|jr|td||jr=|jr=tj |j j |j |j |jd|jid}t|td|n |jr]|j j|j |j |jd|j j|j |j ddtd |d|_dS) Nrz%s: completing multipart uploadParts)rrr@MultipartUploadz%s: completed multipart uploadrrr@rrrrz-%s: wrote 0 bytes to imitate multipart upload)r$rrrHr_upload_next_partrIrGr;r<rcomplete_multipart_uploadrrrKrFr1abort_multipart_upload put_object)rr<rrrrs:       zMultipartWriter.closecCrr7)rGrrrrr*rzMultipartWriter.closedcCr)z+Return True if the stream supports writing.Trrrrrwritable.rzMultipartWriter.writablecCr)zIf False, seek(), tell() and truncate() will raise IOError. We offer only tell support, and no seek or truncate support.Trrrrrr%2r&zMultipartWriter.seekablecCr+r,r-rrrrrrr8r*zMultipartWriter.seekcCr+r,r-rrrrr/<r*zMultipartWriter.truncatecCr))z#Return the current stream position.)rIrrrrr@r*zMultipartWriter.tellcCs td)Nzdetach() not supportedr-rrrrr0Gs zMultipartWriter.detachrr rrVcCsd}t|}|jt|7_|t|krK|}||j|j}|j||||j|jkr?|t|ks;Jt|S||}|t|kst|S)aJWrite the given buffer (bytes, bytearray, memoryview or any buffer interface implementation) to the S3 file. For more information about buffers, see https://docs.python.org/3/c-api/buffer.html There's buffering happening under the covers, so this may not actually do any HTTP transfer right away.r) memoryviewrIrrDrHrrrT)rrrmvrendrrrrJs  zMultipartWriter.writecCsD|jrdStd||jj|j|j|jdd|_td|dS)z'Cancel the underlying multipart upload.Nz %s: terminating multipart uploadrRz%s: terminated multipart upload)rr$rrrVrrrGrrrrr1ks zMultipartWriter.terminatecCs|sJd||j|jS)r2r3)r4rr)rrsrrrr7xs zMultipartWriter.to_boto3Nonec Cs|jd}td|||j|jd|jdtt j |j j |j |j|j||jd}|j|d|dtd|||jd7_|jd|jddS) Nrz3%s: uploading part_num: %i, %i bytes (total %.3fGB)gAr)rrr@ PartNumberrETag)r_r^z#%s: upload of part_num #%i finished)rJr$inforHrrIrrFr1r;r<r upload_partrrrGrKappendrr/)rpart_numuploadrrrrTs0    z!MultipartWriter._upload_next_partcC|Sr7rrrrr __enter__rOzMultipartWriter.__enter__cC |dur |dS|dSr7r1rrexc_typeexc_valexc_tbrrr__exit__  zMultipartWriter.__exit__cCr)Nz%smart_open.s3.MultipartWriter(%r, %r)rrrrrrrzMultipartWriter.__str__cCsd|j|j|jfS)Nz>smart_open.s3.MultipartWriter(bucket=%r, key=%r, part_size=%r))rrrDrrrrr<s zMultipartWriter.__repr__)rr?r7)rr rrV)rr])r2r3r4r@rGDEFAULT_PART_SIZErrNrrrrXr%rrrr/rr0rr1r7rTrfrmrr<rrrrrs4 +(   ! $ rc@seZdZdZdZ   d(ddZddZddZed d Z d d Z d dZ ddZ e jfddZd)ddZddZddZd*ddZd*ddZddZd d!Zd"d#Zd$d%Zd&d'ZdS)+raWrites bytes to S3 using the single part API. Implements the io.BufferedIOBase interface of the standard library. This class buffers all of its input in memory until its `close` method is called. Only then will the data be written to S3 and the buffer is released.NcCs@t||||||durt|_dS|std||_dS)Nz writebuffer needs to be seekable)rrrrHr%r)rrrr8rbrrrrrs  zSinglepartWriter.__init__cCrMr7rrrrrrNrOzSinglepartWriter.flushc Cstd|jr dS|dz3z|jj|j|j|jdWnt j j y4}zt d|j|d}~wwtd|W|j dS|j w)NrrrSz8the bucket %r does not exist, or is forbidden for accessz%s: direct upload finished)r$rrrrrWrrrHrr8r)rr)rerrrrs(   zSinglepartWriter.closecCs|jdup|jjSr7)rHrrrrrrszSinglepartWriter.closedcC |jSz Propagate.)rHrrrrrrrzSinglepartWriter.readablecCrqrr)rHrXrrrrrXrzSinglepartWriter.writablecCrqrr)rHr%rrrrr%rzSinglepartWriter.seekablecCs|j||Srr)rHrrYrrrrszSinglepartWriter.seekcC |j|Srr)rHr/rrrrr/rzSinglepartWriter.truncatecCrqrr)rHrrrrrrrzSinglepartWriter.tellcCrs)a(Write the given buffer (bytes, bytearray, memoryview or any buffer interface implementation) into the buffer. Content of the buffer will be written to S3 on close as a single-part upload. For more information about buffers, see https://docs.python.org/3/c-api/buffer.html)rHr)rrrrrr s zSinglepartWriter.writercCrsrr)rHrrrrrrrzSinglepartWriter.readcCrsrr)rHrrrrrrrzSinglepartWriter.read1cCs|jtd|dS)zClose buffer and skip upload.z %s: terminated singlepart uploadN)rHrr$rrrrrr1s zSinglepartWriter.terminatecCrer7rrrrrrf$rOzSinglepartWriter.__enter__cCrgr7rhrirrrrm'rnzSinglepartWriter.__exit__cCr)Nz&smart_open.s3.SinglepartWriter(%r, %r)rrrrrr-rzSinglepartWriter.__str__cCr)Nz1smart_open.s3.SinglepartWriter(bucket=%r, key=%r)rrrrrr<0rzSinglepartWriter.__repr__)NNNr7r)r2r3r4r@rHrrNrrrrrXr%rrrr/rrrrr1rfrmrr<rrrrrs2      rcCrNTr)rrrr _accept_all4rOrurrc ks|durt}z|j}Wn tyYnw|durtdd\} } tjjdi|} |dur1i}d|vrItjj ||dd|ddd d |d<| jdi|} tj j j |d } t |||| d }tjt||| | d}t||}tj|>}|||}t|ddD])\} \}}|durq| ddkrtd| |t|| d||fV| t|7} qWdn1swYtd| | ddS)a Iterate and download all S3 objects under `s3://bucket_name/prefix`. Parameters ---------- bucket_name: str The name of the bucket. prefix: str, optional Limits the iteration to keys starting with the prefix. accept_key: callable, optional This is a function that accepts a key name (unicode string) and returns True/False, signalling whether the given key should be downloaded. The default behavior is to accept all keys. key_limit: int, optional If specified, the iterator will stop after yielding this many results. workers: int, optional The number of objects to download concurrently. The entire operation uses a single ThreadPoolExecutor and shared thread-safe boto3 S3.Client. Default: 16 retries: int, optional The number of time to retry a failed download. Default: 3 max_threads_per_fileobj: int, optional The maximum number of download threads per worker. The maximum size of the connection pool will be `workers * max_threads_per_fileobj + 1`. Default: 4 client_kwargs: dict, optional Keyword arguments to pass when creating a new session. For a list of available names and values, see: https://boto3.amazonaws.com/v1/documentation/api/latest/reference/core/session.html#boto3.session.Session.client session_kwargs: dict, optional Keyword arguments to pass when creating a new session. For a list of available names and values, see: https://boto3.amazonaws.com/v1/documentation/api/latest/reference/core/session.html#boto3.session.Session Yields ------ str The full key name (does not include the bucket name). bytes The full contents of the key. Notes ----- The keys are processed in parallel, using `workers` processes (default: 16), to speed up downloads greatly. If multiprocessing is not available, thus _MULTIPROCESSING is False, this parameter will be ignored. Examples -------- >>> # get all JSON files under "mybucket/foo/" >>> for key, content in iter_bucket( ... bucket_name, prefix='foo/', ... accept_key=lambda key: key.endswith('.json')): ... print key, len(content) >>> # limit to 10k files, using 32 parallel workers (default is 16) >>> for key, content in iter_bucket(bucket_name, key_limit=10000, workers=32): ... print key, len(content) Nzbucket_name may not be None)rrrrTrrrrr )max_concurrency) bucket_nameprefix accept_keyr8)ryrr8transfer_config)rirz-yielding key #%i: %s, size %i (total %.1f MB)g0Az%processed %i keys, total size %.1f MBrr)rurrrrruSessionrr8rr transferTransferConfig _list_bucketr;r< _download_key itertoolsislicerM concurrencyThreadPoolExecutorimap enumerater$r`r)ryrzr{ key_limitworkersrmax_threads_per_fileobjrbsession_kwargs total_sizekey_norur8r| key_iterator download_keyexecutorresult_iteratorrcontentrrr iter_bucket8snF         rcCrrtr)rxrrrsr)rzr{c csd} |rt|||d}nt||d}|jdi|}z|d}Wn ty+Ynw|D] }|d} || r;| Vq.|dd}|sFdSq)NT)rPrefixContinuationToken)rrContentsrNextContinuationTokenr)rWlist_objects_v2rdr+) ryr8rzr{ctokenr9r*rcrrrrrs*    rc Cst|dD]B}z t||||d}Wn0tjjyB}z"d|jvr3|jdddkr3WYd}~dS||kr8WYd}~qd}~ww||fSdS)Nr)r8rykey_namer|r r!404rA)r"_download_fileobjrrr)r*r+)rr8ryrr|x content_bytesr/rrrrs$   rcCs"t}|j||||d|S)N)rrFileobjr)rrdownload_fileobjr)r8ryrr|bufrrrrsr)rvNNrrwrN)Cr@ __future__rrlrr;rloggingr'rzmathrtypingrrrrbotocore.clientrbotocore.exceptionsurllib3.exceptionsr ImportErrorr5smart_open.bytebufferrMsmart_open.concurrencysmart_open.utilsrmypy_boto3_s3.clientrtyping_extensionsr getLoggerr2r$rBrorCrPrQrRr= URI_EXAMPLESrrrrFr6rarprr~rrobjectrrBufferedIOBaserrrrurrrrrrrrs         'G=& Xqy