| 12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583158415851586158715881589159015911592159315941595159615971598159916001601160216031604160516061607160816091610161116121613161416151616161716181619162016211622162316241625162616271628162916301631163216331634163516361637163816391640164116421643164416451646164716481649165016511652165316541655165616571658165916601661166216631664166516661667166816691670167116721673167416751676167716781679168016811682168316841685168616871688168916901691169216931694169516961697169816991700170117021703170417051706170717081709171017111712171317141715171617171718171917201721172217231724172517261727172817291730173117321733173417351736173717381739174017411742174317441745174617471748174917501751175217531754175517561757175817591760176117621763176417651766176717681769177017711772177317741775177617771778177917801781178217831784178517861787178817891790179117921793179417951796179717981799180018011802180318041805180618071808180918101811181218131814181518161817181818191820182118221823182418251826182718281829183018311832183318341835183618371838183918401841184218431844184518461847184818491850185118521853185418551856185718581859186018611862186318641865186618671868186918701871187218731874187518761877187818791880188118821883188418851886188718881889189018911892189318941895189618971898189919001901190219031904190519061907190819091910191119121913191419151916191719181919192019211922192319241925192619271928192919301931193219331934193519361937193819391940194119421943194419451946194719481949195019511952195319541955195619571958195919601961196219631964196519661967196819691970197119721973197419751976197719781979198019811982198319841985198619871988198919901991199219931994199519961997199819992000200120022003200420052006200720082009201020112012201320142015201620172018201920202021202220232024202520262027202820292030203120322033203420352036203720382039204020412042204320442045204620472048204920502051205220532054205520562057205820592060206120622063206420652066206720682069207020712072207320742075207620772078207920802081208220832084208520862087208820892090209120922093209420952096209720982099210021012102210321042105210621072108210921102111211221132114211521162117211821192120212121222123212421252126212721282129213021312132213321342135213621372138213921402141214221432144214521462147214821492150215121522153215421552156215721582159216021612162216321642165216621672168216921702171217221732174217521762177217821792180218121822183218421852186218721882189219021912192219321942195219621972198219922002201220222032204220522062207220822092210221122122213221422152216221722182219222022212222222322242225222622272228222922302231223222332234223522362237223822392240224122422243224422452246224722482249225022512252225322542255225622572258225922602261226222632264226522662267226822692270227122722273227422752276227722782279228022812282228322842285228622872288228922902291229222932294229522962297229822992300230123022303230423052306230723082309231023112312231323142315231623172318231923202321232223232324232523262327232823292330233123322333233423352336233723382339234023412342234323442345234623472348234923502351235223532354235523562357235823592360236123622363236423652366236723682369237023712372237323742375237623772378237923802381238223832384238523862387238823892390239123922393239423952396239723982399240024012402240324042405240624072408240924102411241224132414241524162417241824192420242124222423242424252426242724282429243024312432243324342435243624372438243924402441244224432444244524462447244824492450245124522453245424552456245724582459246024612462246324642465246624672468246924702471247224732474247524762477247824792480248124822483248424852486248724882489249024912492249324942495249624972498249925002501250225032504250525062507250825092510251125122513251425152516251725182519252025212522252325242525252625272528252925302531253225332534253525362537253825392540254125422543254425452546254725482549255025512552255325542555255625572558255925602561256225632564256525662567256825692570257125722573257425752576257725782579258025812582258325842585258625872588258925902591259225932594259525962597259825992600260126022603260426052606260726082609261026112612261326142615261626172618261926202621262226232624262526262627262826292630263126322633263426352636263726382639264026412642264326442645264626472648264926502651265226532654265526562657265826592660266126622663266426652666266726682669267026712672267326742675267626772678267926802681268226832684268526862687268826892690269126922693269426952696269726982699270027012702270327042705270627072708270927102711271227132714271527162717271827192720272127222723272427252726272727282729273027312732273327342735273627372738273927402741274227432744274527462747274827492750275127522753275427552756275727582759276027612762276327642765276627672768276927702771277227732774277527762777277827792780278127822783278427852786278727882789279027912792279327942795279627972798279928002801280228032804280528062807280828092810281128122813281428152816281728182819282028212822282328242825282628272828282928302831283228332834283528362837283828392840284128422843284428452846284728482849285028512852285328542855285628572858285928602861286228632864286528662867286828692870287128722873287428752876287728782879288028812882288328842885288628872888288928902891289228932894289528962897289828992900290129022903290429052906290729082909291029112912291329142915291629172918291929202921292229232924292529262927292829292930293129322933293429352936293729382939294029412942294329442945294629472948294929502951295229532954295529562957295829592960296129622963296429652966296729682969297029712972297329742975297629772978297929802981298229832984298529862987298829892990299129922993299429952996299729982999300030013002300330043005300630073008300930103011301230133014301530163017301830193020302130223023302430253026302730283029303030313032303330343035303630373038303930403041304230433044304530463047304830493050305130523053305430553056305730583059306030613062306330643065306630673068306930703071307230733074307530763077307830793080308130823083308430853086308730883089309030913092309330943095309630973098309931003101310231033104310531063107310831093110311131123113311431153116311731183119312031213122312331243125312631273128312931303131313231333134313531363137313831393140314131423143314431453146314731483149315031513152315331543155315631573158315931603161316231633164316531663167316831693170317131723173317431753176317731783179318031813182318331843185318631873188318931903191319231933194319531963197319831993200320132023203320432053206320732083209321032113212321332143215321632173218321932203221322232233224322532263227322832293230323132323233323432353236323732383239324032413242324332443245324632473248324932503251325232533254325532563257325832593260326132623263326432653266326732683269327032713272327332743275327632773278327932803281328232833284328532863287328832893290329132923293329432953296329732983299330033013302330333043305330633073308330933103311331233133314331533163317331833193320332133223323332433253326332733283329333033313332333333343335333633373338333933403341334233433344334533463347334833493350335133523353335433553356335733583359336033613362336333643365336633673368336933703371337233733374337533763377337833793380338133823383338433853386338733883389339033913392339333943395339633973398339934003401340234033404340534063407340834093410341134123413341434153416341734183419342034213422342334243425342634273428342934303431343234333434343534363437343834393440344134423443 |
- # -*- coding: utf-8 -*-
- # MinIO Python Library for Amazon S3 Compatible Cloud Storage, (C)
- # 2015, 2016, 2017 MinIO, Inc.
- #
- # Licensed under the Apache License, Version 2.0 (the "License");
- # you may not use this file except in compliance with the License.
- # You may obtain a copy of the License at
- #
- # http://www.apache.org/licenses/LICENSE-2.0
- #
- # Unless required by applicable law or agreed to in writing, software
- # distributed under the License is distributed on an "AS IS" BASIS,
- # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- # See the License for the specific language governing permissions and
- # limitations under the License.
- # pylint: disable=too-many-arguments
- # pylint: disable=too-many-branches
- # pylint: disable=too-many-function-args
- # pylint: disable=too-many-lines
- # pylint: disable=too-many-public-methods
- # pylint: disable=too-many-statements
- # pylint: disable=too-many-locals
- # pylint: disable=too-many-positional-arguments
- """
- Simple Storage Service (aka S3) client to perform bucket and object operations.
- """
- from __future__ import absolute_import, annotations
- import itertools
- import json
- import os
- import tarfile
- from collections.abc import Iterable
- from datetime import datetime, timedelta
- from io import BytesIO
- from random import random
- from typing import (Any, BinaryIO, Iterator, Optional, TextIO, Tuple, Union,
- cast)
- from urllib.parse import urlunsplit
- from xml.etree import ElementTree as ET
- import certifi
- import urllib3
- from urllib3 import Retry
- from urllib3._collections import HTTPHeaderDict
- try:
- from urllib3.response import BaseHTTPResponse # type: ignore[attr-defined]
- except ImportError:
- from urllib3.response import HTTPResponse as BaseHTTPResponse
- from urllib3.util import Timeout
- from . import time
- from .commonconfig import (COPY, REPLACE, ComposeSource, CopySource,
- SnowballObject, Tags)
- from .credentials import Credentials, StaticProvider
- from .credentials.providers import Provider
- from .datatypes import (Bucket, CompleteMultipartUploadResult, EventIterable,
- ListAllMyBucketsResult, ListMultipartUploadsResult,
- ListPartsResult, Object, Part, PostPolicy,
- parse_copy_object, parse_list_objects)
- from .deleteobjects import (DeleteError, DeleteObject, DeleteRequest,
- DeleteResult)
- from .error import InvalidResponseError, S3Error, ServerError
- from .helpers import (_DEFAULT_USER_AGENT, MAX_MULTIPART_COUNT,
- MAX_MULTIPART_OBJECT_SIZE, MAX_PART_SIZE, MIN_PART_SIZE,
- BaseURL, DictType, ObjectWriteResult, ProgressType,
- ThreadPool, check_bucket_name, check_object_name,
- check_sse, check_ssec, genheaders, get_part_info,
- headers_to_strings, is_valid_policy_type, makedirs,
- md5sum_hash, queryencode, read_part_data, sha256_hash)
- from .legalhold import LegalHold
- from .lifecycleconfig import LifecycleConfig
- from .notificationconfig import NotificationConfig
- from .objectlockconfig import ObjectLockConfig
- from .replicationconfig import ReplicationConfig
- from .retention import Retention
- from .select import SelectObjectReader, SelectRequest
- from .signer import presign_v4, sign_v4_s3
- from .sse import Sse, SseCustomerKey
- from .sseconfig import SSEConfig
- from .tagging import Tagging
- from .versioningconfig import VersioningConfig
- from .xml import Element, SubElement, findtext, getbytes, marshal, unmarshal
- class Minio:
- """
- Simple Storage Service (aka S3) client to perform bucket and object
- operations.
- :param endpoint: Hostname of a S3 service.
- :param access_key: Access key (aka user ID) of your account in S3 service.
- :param secret_key: Secret Key (aka password) of your account in S3 service.
- :param session_token: Session token of your account in S3 service.
- :param secure: Flag to indicate to use secure (TLS) connection to S3
- service or not.
- :param region: Region name of buckets in S3 service.
- :param http_client: Customized HTTP client.
- :param credentials: Credentials provider of your account in S3 service.
- :param cert_check: Flag to indicate to verify SSL certificate or not.
- :return: :class:`Minio <Minio>` object
- Example::
- # Create client with anonymous access.
- client = Minio("play.min.io")
- # Create client with access and secret key.
- client = Minio("s3.amazonaws.com", "ACCESS-KEY", "SECRET-KEY")
- # Create client with access key and secret key with specific region.
- client = Minio(
- "play.minio.io:9000",
- access_key="Q3AM3UQ867SPQQA43P2F",
- secret_key="zuf+tfteSlswRu7BJ86wekitnifILbZam1KYY3TG",
- region="my-region",
- )
- **NOTE on concurrent usage:** `Minio` object is thread safe when using
- the Python `threading` library. Specifically, it is **NOT** safe to share
- it between multiple processes, for example when using
- `multiprocessing.Pool`. The solution is simply to create a new `Minio`
- object in each process, and not share it between processes.
- """
- _region_map: dict[str, str]
- _base_url: BaseURL
- _user_agent: str
- _trace_stream: Optional[TextIO]
- _provider: Optional[Provider]
- _http: urllib3.PoolManager
- def __init__(
- self,
- endpoint: str,
- access_key: Optional[str] = None,
- secret_key: Optional[str] = None,
- session_token: Optional[str] = None,
- secure: bool = True,
- region: Optional[str] = None,
- http_client: Optional[urllib3.PoolManager] = None,
- credentials: Optional[Provider] = None,
- cert_check: bool = True,
- ):
- # Validate http client has correct base class.
- if http_client and not isinstance(http_client, urllib3.PoolManager):
- raise ValueError(
- "HTTP client should be instance of `urllib3.PoolManager`"
- )
- self._region_map = {}
- self._base_url = BaseURL(
- ("https://" if secure else "http://") + endpoint,
- region,
- )
- self._user_agent = _DEFAULT_USER_AGENT
- self._trace_stream = None
- if access_key:
- if secret_key is None:
- raise ValueError("secret key must be provided with access key")
- credentials = StaticProvider(access_key, secret_key, session_token)
- self._provider = credentials
- # Load CA certificates from SSL_CERT_FILE file if set
- timeout = timedelta(minutes=5).seconds
- self._http = http_client or urllib3.PoolManager(
- timeout=Timeout(connect=timeout, read=timeout),
- maxsize=10,
- cert_reqs='CERT_REQUIRED' if cert_check else 'CERT_NONE',
- ca_certs=os.environ.get('SSL_CERT_FILE') or certifi.where(),
- retries=Retry(
- total=5,
- backoff_factor=0.2,
- status_forcelist=[500, 502, 503, 504]
- )
- )
- def __del__(self):
- if hasattr(self, "_http"): # Only required for unit test run
- self._http.clear()
- def _handle_redirect_response(
- self,
- method: str,
- response: BaseHTTPResponse,
- bucket_name: Optional[str] = None,
- retry: bool = False,
- ) -> tuple[Optional[str], Optional[str]]:
- """
- Handle redirect response indicates whether retry HEAD request
- on failure.
- """
- code, message = {
- 301: ("PermanentRedirect", "Moved Permanently"),
- 307: ("Redirect", "Temporary redirect"),
- 400: ("BadRequest", "Bad request"),
- }.get(response.status, (None, None))
- region = response.headers.get("x-amz-bucket-region")
- if message and region:
- message += "; use region " + region
- if (
- retry and region and method == "HEAD" and bucket_name and
- self._region_map.get(bucket_name)
- ):
- code, message = ("RetryHead", None)
- return code, message
- def _build_headers(
- self,
- host: str,
- headers: Optional[DictType] = None,
- body: Optional[bytes] = None,
- creds: Optional[Credentials] = None,
- ) -> tuple[DictType, datetime]:
- """Build headers with given parameters."""
- headers = headers or {}
- md5sum_added = headers.get("Content-MD5")
- headers["Host"] = host
- headers["User-Agent"] = self._user_agent
- sha256 = None
- md5sum = None
- if body:
- headers["Content-Length"] = str(len(body))
- if creds:
- if self._base_url.is_https:
- sha256 = "UNSIGNED-PAYLOAD"
- md5sum = None if md5sum_added else md5sum_hash(body)
- else:
- sha256 = sha256_hash(body)
- else:
- md5sum = None if md5sum_added else md5sum_hash(body)
- if md5sum:
- headers["Content-MD5"] = md5sum
- if sha256:
- headers["x-amz-content-sha256"] = sha256
- if creds and creds.session_token:
- headers["X-Amz-Security-Token"] = creds.session_token
- date = time.utcnow()
- headers["x-amz-date"] = time.to_amz_date(date)
- return headers, date
- def _url_open(
- self,
- method: str,
- region: str,
- bucket_name: Optional[str] = None,
- object_name: Optional[str] = None,
- body: Optional[bytes] = None,
- headers: Optional[DictType] = None,
- query_params: Optional[DictType] = None,
- preload_content: bool = True,
- no_body_trace: bool = False,
- ) -> BaseHTTPResponse:
- """Execute HTTP request."""
- creds = self._provider.retrieve() if self._provider else None
- url = self._base_url.build(
- method=method,
- region=region,
- bucket_name=bucket_name,
- object_name=object_name,
- query_params=query_params,
- )
- headers, date = self._build_headers(url.netloc, headers, body, creds)
- if creds:
- headers = sign_v4_s3(
- method=method,
- url=url,
- region=region,
- headers=headers,
- credentials=creds,
- content_sha256=cast(str, headers.get("x-amz-content-sha256")),
- date=date,
- )
- if self._trace_stream:
- self._trace_stream.write("---------START-HTTP---------\n")
- query = ("?" + url.query) if url.query else ""
- self._trace_stream.write(f"{method} {url.path}{query} HTTP/1.1\n")
- self._trace_stream.write(
- headers_to_strings(headers, titled_key=True),
- )
- self._trace_stream.write("\n")
- if not no_body_trace and body is not None:
- self._trace_stream.write("\n")
- self._trace_stream.write(
- body.decode() if isinstance(body, bytes) else str(body),
- )
- self._trace_stream.write("\n")
- self._trace_stream.write("\n")
- http_headers = HTTPHeaderDict()
- for key, value in (headers or {}).items():
- if isinstance(value, (list, tuple)):
- for val in value:
- http_headers.add(key, val)
- else:
- http_headers.add(key, value)
- response = self._http.urlopen(
- method,
- urlunsplit(url),
- body=body,
- headers=http_headers,
- preload_content=preload_content,
- )
- if self._trace_stream:
- self._trace_stream.write(f"HTTP/1.1 {response.status}\n")
- self._trace_stream.write(
- headers_to_strings(response.headers),
- )
- self._trace_stream.write("\n")
- if response.status in [200, 204, 206]:
- if self._trace_stream:
- if preload_content:
- self._trace_stream.write("\n")
- self._trace_stream.write(response.data.decode())
- self._trace_stream.write("\n")
- self._trace_stream.write("----------END-HTTP----------\n")
- return response
- response.read(cache_content=True)
- if not preload_content:
- response.release_conn()
- if self._trace_stream and method != "HEAD" and response.data:
- self._trace_stream.write(response.data.decode())
- self._trace_stream.write("\n")
- if (
- method != "HEAD" and
- "application/xml" not in response.headers.get(
- "content-type", "",
- ).split(";")
- ):
- if self._trace_stream:
- self._trace_stream.write("----------END-HTTP----------\n")
- if response.status == 304 and not response.data:
- raise ServerError(
- f"server failed with HTTP status code {response.status}",
- response.status,
- )
- raise InvalidResponseError(
- response.status,
- cast(str, response.headers.get("content-type")),
- response.data.decode() if response.data else None,
- )
- if not response.data and method != "HEAD":
- if self._trace_stream:
- self._trace_stream.write("----------END-HTTP----------\n")
- raise InvalidResponseError(
- response.status,
- response.headers.get("content-type"),
- None,
- )
- response_error = S3Error.fromxml(response) if response.data else None
- if self._trace_stream:
- self._trace_stream.write("----------END-HTTP----------\n")
- error_map = {
- 301: lambda: self._handle_redirect_response(
- method, response, bucket_name, True,
- ),
- 307: lambda: self._handle_redirect_response(
- method, response, bucket_name, True,
- ),
- 400: lambda: self._handle_redirect_response(
- method, response, bucket_name, True,
- ),
- 403: lambda: ("AccessDenied", "Access denied"),
- 404: lambda: (
- ("NoSuchKey", "Object does not exist")
- if object_name
- else ("NoSuchBucket", "Bucket does not exist")
- if bucket_name
- else ("ResourceNotFound", "Request resource not found")
- ),
- 405: lambda: (
- "MethodNotAllowed",
- "The specified method is not allowed against this resource",
- ),
- 409: lambda: (
- ("NoSuchBucket", "Bucket does not exist")
- if bucket_name
- else ("ResourceConflict", "Request resource conflicts"),
- ),
- 501: lambda: (
- "MethodNotAllowed",
- "The specified method is not allowed against this resource",
- ),
- }
- if not response_error:
- func = error_map.get(response.status)
- code, message = func() if func else (None, None)
- if not code:
- raise ServerError(
- f"server failed with HTTP status code {response.status}",
- response.status,
- )
- response_error = S3Error(
- response=response,
- code=cast(str, code),
- message=cast(Union[str, None], message),
- resource=url.path,
- request_id=response.headers.get("x-amz-request-id"),
- host_id=response.headers.get("x-amz-id-2"),
- bucket_name=bucket_name,
- object_name=object_name,
- )
- if response_error.code in ["NoSuchBucket", "RetryHead"]:
- if bucket_name is not None:
- self._region_map.pop(bucket_name, None)
- raise response_error
- def _execute(
- self,
- method: str,
- bucket_name: Optional[str] = None,
- object_name: Optional[str] = None,
- body: Optional[bytes] = None,
- headers: Optional[DictType] = None,
- query_params: Optional[DictType] = None,
- preload_content: bool = True,
- no_body_trace: bool = False,
- ) -> BaseHTTPResponse:
- """Execute HTTP request."""
- region = self._get_region(bucket_name)
- try:
- return self._url_open(
- method,
- region,
- bucket_name=bucket_name,
- object_name=object_name,
- body=body,
- headers=headers,
- query_params=query_params,
- preload_content=preload_content,
- no_body_trace=no_body_trace,
- )
- except S3Error as exc:
- if exc.code != "RetryHead":
- raise
- # Retry only once on RetryHead error.
- try:
- return self._url_open(
- method,
- region,
- bucket_name=bucket_name,
- object_name=object_name,
- body=body,
- headers=headers,
- query_params=query_params,
- preload_content=preload_content,
- no_body_trace=no_body_trace,
- )
- except S3Error as exc:
- if exc.code != "RetryHead":
- raise
- code, message = self._handle_redirect_response(
- method, exc.response, bucket_name,
- )
- raise exc.copy(cast(str, code), cast(str, message))
- def _get_region(self, bucket_name: Optional[str] = None) -> str:
- """
- Return region of given bucket either from region cache or set in
- constructor.
- """
- if self._base_url.region:
- return self._base_url.region
- if not bucket_name or not self._provider:
- return "us-east-1"
- region = self._region_map.get(bucket_name)
- if region:
- return region
- # Execute GetBucketLocation REST API to get region of the bucket.
- response = self._url_open(
- "GET",
- "us-east-1",
- bucket_name=bucket_name,
- query_params={"location": ""},
- )
- element = ET.fromstring(response.data.decode())
- if not element.text:
- region = "us-east-1"
- elif element.text == "EU" and self._base_url.is_aws_host:
- region = "eu-west-1"
- else:
- region = element.text
- self._region_map[bucket_name] = region
- return region
- def set_app_info(self, app_name: str, app_version: str):
- """
- Set your application name and version to user agent header.
- :param app_name: Application name.
- :param app_version: Application version.
- Example::
- client.set_app_info('my_app', '1.0.2')
- """
- if not (app_name and app_version):
- raise ValueError("Application name/version cannot be empty.")
- self._user_agent = f"{_DEFAULT_USER_AGENT} {app_name}/{app_version}"
- def trace_on(self, stream: TextIO):
- """
- Enable http trace.
- :param stream: Stream for writing HTTP call tracing.
- """
- if not stream:
- raise ValueError('Input stream for trace output is invalid.')
- # Save new output stream.
- self._trace_stream = stream
- def trace_off(self):
- """
- Disable HTTP trace.
- """
- self._trace_stream = None
- def enable_accelerate_endpoint(self):
- """Enables accelerate endpoint for Amazon S3 endpoint."""
- self._base_url.accelerate_host_flag = True
- def disable_accelerate_endpoint(self):
- """Disables accelerate endpoint for Amazon S3 endpoint."""
- self._base_url.accelerate_host_flag = False
- def enable_dualstack_endpoint(self):
- """Enables dualstack endpoint for Amazon S3 endpoint."""
- self._base_url.dualstack_host_flag = True
- def disable_dualstack_endpoint(self):
- """Disables dualstack endpoint for Amazon S3 endpoint."""
- self._base_url.dualstack_host_flag = False
- def enable_virtual_style_endpoint(self):
- """Enables virtual style endpoint."""
- self._base_url.virtual_style_flag = True
- def disable_virtual_style_endpoint(self):
- """Disables virtual style endpoint."""
- self._base_url.virtual_style_flag = False
- def select_object_content(
- self,
- bucket_name: str,
- object_name: str,
- request: SelectRequest,
- ) -> SelectObjectReader:
- """
- Select content of an object by SQL expression.
- :param bucket_name: Name of the bucket.
- :param object_name: Object name in the bucket.
- :param request: :class:`SelectRequest <SelectRequest>` object.
- :return: A reader contains requested records and progress information.
- Example::
- with client.select_object_content(
- "my-bucket",
- "my-object.csv",
- SelectRequest(
- "select * from S3Object",
- CSVInputSerialization(),
- CSVOutputSerialization(),
- request_progress=True,
- ),
- ) as result:
- for data in result.stream():
- print(data.decode())
- print(result.stats())
- """
- check_bucket_name(bucket_name, s3_check=self._base_url.is_aws_host)
- check_object_name(object_name)
- if not isinstance(request, SelectRequest):
- raise ValueError("request must be SelectRequest type")
- body = marshal(request)
- response = self._execute(
- "POST",
- bucket_name=bucket_name,
- object_name=object_name,
- body=body,
- headers={"Content-MD5": cast(str, md5sum_hash(body))},
- query_params={"select": "", "select-type": "2"},
- preload_content=False,
- )
- return SelectObjectReader(response)
- def make_bucket(
- self,
- bucket_name: str,
- location: Optional[str] = None,
- object_lock: bool = False,
- ):
- """
- Create a bucket with region and object lock.
- :param bucket_name: Name of the bucket.
- :param location: Region in which the bucket will be created.
- :param object_lock: Flag to set object-lock feature.
- Examples::
- # Create bucket.
- client.make_bucket("my-bucket")
- # Create bucket on specific region.
- client.make_bucket("my-bucket", "us-west-1")
- # Create bucket with object-lock feature on specific region.
- client.make_bucket("my-bucket", "eu-west-2", object_lock=True)
- """
- check_bucket_name(bucket_name, True,
- s3_check=self._base_url.is_aws_host)
- if self._base_url.region:
- # Error out if region does not match with region passed via
- # constructor.
- if location and self._base_url.region != location:
- raise ValueError(
- f"region must be {self._base_url.region}, "
- f"but passed {location}"
- )
- location = self._base_url.region or location or "us-east-1"
- headers: Optional[DictType] = (
- {"x-amz-bucket-object-lock-enabled": "true"}
- if object_lock else None
- )
- body = None
- if location != "us-east-1":
- element = Element("CreateBucketConfiguration")
- SubElement(element, "LocationConstraint", location)
- body = getbytes(element)
- self._url_open(
- "PUT",
- location,
- bucket_name=bucket_name,
- body=body,
- headers=headers,
- )
- self._region_map[bucket_name] = location
- def list_buckets(self) -> list[Bucket]:
- """
- List information of all accessible buckets.
- :return: List of :class:`Bucket <Bucket>` object.
- Example::
- buckets = client.list_buckets()
- for bucket in buckets:
- print(bucket.name, bucket.creation_date)
- """
- response = self._execute("GET")
- result = unmarshal(ListAllMyBucketsResult, response.data.decode())
- return result.buckets
- def bucket_exists(self, bucket_name: str) -> bool:
- """
- Check if a bucket exists.
- :param bucket_name: Name of the bucket.
- :return: True if the bucket exists.
- Example::
- if client.bucket_exists("my-bucket"):
- print("my-bucket exists")
- else:
- print("my-bucket does not exist")
- """
- check_bucket_name(bucket_name, s3_check=self._base_url.is_aws_host)
- try:
- self._execute("HEAD", bucket_name)
- return True
- except S3Error as exc:
- if exc.code != "NoSuchBucket":
- raise
- return False
- def remove_bucket(self, bucket_name: str):
- """
- Remove an empty bucket.
- :param bucket_name: Name of the bucket.
- Example::
- client.remove_bucket("my-bucket")
- """
- check_bucket_name(bucket_name, s3_check=self._base_url.is_aws_host)
- self._execute("DELETE", bucket_name)
- self._region_map.pop(bucket_name, None)
- def get_bucket_policy(self, bucket_name: str) -> str:
- """
- Get bucket policy configuration of a bucket.
- :param bucket_name: Name of the bucket.
- :return: Bucket policy configuration as JSON string.
- Example::
- policy = client.get_bucket_policy("my-bucket")
- """
- check_bucket_name(bucket_name, s3_check=self._base_url.is_aws_host)
- response = self._execute(
- "GET", bucket_name, query_params={"policy": ""},
- )
- return response.data.decode()
- def delete_bucket_policy(self, bucket_name: str):
- """
- Delete bucket policy configuration of a bucket.
- :param bucket_name: Name of the bucket.
- Example::
- client.delete_bucket_policy("my-bucket")
- """
- check_bucket_name(bucket_name, s3_check=self._base_url.is_aws_host)
- self._execute("DELETE", bucket_name, query_params={"policy": ""})
- def set_bucket_policy(self, bucket_name: str, policy: str | bytes):
- """
- Set bucket policy configuration to a bucket.
- :param bucket_name: Name of the bucket.
- :param policy: Bucket policy configuration as JSON string.
- Example::
- client.set_bucket_policy("my-bucket", policy)
- """
- check_bucket_name(bucket_name, s3_check=self._base_url.is_aws_host)
- is_valid_policy_type(policy)
- self._execute(
- "PUT",
- bucket_name,
- body=policy if isinstance(policy, bytes) else policy.encode(),
- headers={"Content-MD5": cast(str, md5sum_hash(policy))},
- query_params={"policy": ""},
- )
- def get_bucket_notification(self, bucket_name: str) -> NotificationConfig:
- """
- Get notification configuration of a bucket.
- :param bucket_name: Name of the bucket.
- :return: :class:`NotificationConfig <NotificationConfig>` object.
- Example::
- config = client.get_bucket_notification("my-bucket")
- """
- check_bucket_name(bucket_name, s3_check=self._base_url.is_aws_host)
- response = self._execute(
- "GET", bucket_name, query_params={"notification": ""},
- )
- return unmarshal(NotificationConfig, response.data.decode())
- def set_bucket_notification(
- self,
- bucket_name: str,
- config: NotificationConfig,
- ):
- """
- Set notification configuration of a bucket.
- :param bucket_name: Name of the bucket.
- :param config: class:`NotificationConfig <NotificationConfig>` object.
- Example::
- config = NotificationConfig(
- queue_config_list=[
- QueueConfig(
- "QUEUE-ARN-OF-THIS-BUCKET",
- ["s3:ObjectCreated:*"],
- config_id="1",
- prefix_filter_rule=PrefixFilterRule("abc"),
- ),
- ],
- )
- client.set_bucket_notification("my-bucket", config)
- """
- check_bucket_name(bucket_name, s3_check=self._base_url.is_aws_host)
- if not isinstance(config, NotificationConfig):
- raise ValueError("config must be NotificationConfig type")
- body = marshal(config)
- self._execute(
- "PUT",
- bucket_name,
- body=body,
- headers={"Content-MD5": cast(str, md5sum_hash(body))},
- query_params={"notification": ""},
- )
- def delete_bucket_notification(self, bucket_name: str):
- """
- Delete notification configuration of a bucket. On success, S3 service
- stops notification of events previously set of the bucket.
- :param bucket_name: Name of the bucket.
- Example::
- client.delete_bucket_notification("my-bucket")
- """
- self.set_bucket_notification(bucket_name, NotificationConfig())
- def set_bucket_encryption(self, bucket_name: str, config: SSEConfig):
- """
- Set encryption configuration of a bucket.
- :param bucket_name: Name of the bucket.
- :param config: :class:`SSEConfig <SSEConfig>` object.
- Example::
- client.set_bucket_encryption(
- "my-bucket", SSEConfig(Rule.new_sse_s3_rule()),
- )
- """
- check_bucket_name(bucket_name, s3_check=self._base_url.is_aws_host)
- if not isinstance(config, SSEConfig):
- raise ValueError("config must be SSEConfig type")
- body = marshal(config)
- self._execute(
- "PUT",
- bucket_name,
- body=body,
- headers={"Content-MD5": cast(str, md5sum_hash(body))},
- query_params={"encryption": ""},
- )
- def get_bucket_encryption(self, bucket_name: str) -> Optional[SSEConfig]:
- """
- Get encryption configuration of a bucket.
- :param bucket_name: Name of the bucket.
- :return: :class:`SSEConfig <SSEConfig>` object.
- Example::
- config = client.get_bucket_encryption("my-bucket")
- """
- check_bucket_name(bucket_name, s3_check=self._base_url.is_aws_host)
- try:
- response = self._execute(
- "GET",
- bucket_name,
- query_params={"encryption": ""},
- )
- return unmarshal(SSEConfig, response.data.decode())
- except S3Error as exc:
- if exc.code != "ServerSideEncryptionConfigurationNotFoundError":
- raise
- return None
- def delete_bucket_encryption(self, bucket_name: str):
- """
- Delete encryption configuration of a bucket.
- :param bucket_name: Name of the bucket.
- Example::
- client.delete_bucket_encryption("my-bucket")
- """
- check_bucket_name(bucket_name, s3_check=self._base_url.is_aws_host)
- try:
- self._execute(
- "DELETE",
- bucket_name,
- query_params={"encryption": ""},
- )
- except S3Error as exc:
- if exc.code != "ServerSideEncryptionConfigurationNotFoundError":
- raise
- def listen_bucket_notification(
- self,
- bucket_name: str,
- prefix: str = "",
- suffix: str = "",
- events: tuple[str, ...] = ('s3:ObjectCreated:*',
- 's3:ObjectRemoved:*',
- 's3:ObjectAccessed:*'),
- ) -> EventIterable:
- """
- Listen events of object prefix and suffix of a bucket. Caller should
- iterate returned iterator to read new events.
- :param bucket_name: Name of the bucket.
- :param prefix: Listen events of object starts with prefix.
- :param suffix: Listen events of object ends with suffix.
- :param events: Events to listen.
- :return: Iterator of event records as :dict:.
- Example::
- with client.listen_bucket_notification(
- "my-bucket",
- prefix="my-prefix/",
- events=["s3:ObjectCreated:*", "s3:ObjectRemoved:*"],
- ) as events:
- for event in events:
- print(event)
- """
- check_bucket_name(bucket_name, s3_check=self._base_url.is_aws_host)
- if self._base_url.is_aws_host:
- raise ValueError(
- "ListenBucketNotification API is not supported in Amazon S3",
- )
- return EventIterable(
- lambda: self._execute(
- "GET",
- bucket_name,
- query_params={
- "prefix": prefix or "",
- "suffix": suffix or "",
- "events": cast(Tuple[str], events),
- },
- preload_content=False,
- ),
- )
- def set_bucket_versioning(
- self,
- bucket_name: str,
- config: VersioningConfig,
- ):
- """
- Set versioning configuration to a bucket.
- :param bucket_name: Name of the bucket.
- :param config: :class:`VersioningConfig <VersioningConfig>`.
- Example::
- client.set_bucket_versioning(
- "my-bucket", VersioningConfig(ENABLED),
- )
- """
- check_bucket_name(bucket_name, s3_check=self._base_url.is_aws_host)
- if not isinstance(config, VersioningConfig):
- raise ValueError("config must be VersioningConfig type")
- body = marshal(config)
- self._execute(
- "PUT",
- bucket_name,
- body=body,
- headers={"Content-MD5": cast(str, md5sum_hash(body))},
- query_params={"versioning": ""},
- )
- def get_bucket_versioning(self, bucket_name: str) -> VersioningConfig:
- """
- Get versioning configuration of a bucket.
- :param bucket_name: Name of the bucket.
- :return: :class:`VersioningConfig <VersioningConfig>`.
- Example::
- config = client.get_bucket_versioning("my-bucket")
- print(config.status)
- """
- check_bucket_name(bucket_name, s3_check=self._base_url.is_aws_host)
- response = self._execute(
- "GET",
- bucket_name,
- query_params={"versioning": ""},
- )
- return unmarshal(VersioningConfig, response.data.decode())
- def fput_object(
- self,
- bucket_name: str,
- object_name: str,
- file_path: str,
- content_type: str = "application/octet-stream",
- metadata: Optional[DictType] = None,
- sse: Optional[Sse] = None,
- progress: Optional[ProgressType] = None,
- part_size: int = 0,
- num_parallel_uploads: int = 3,
- tags: Optional[Tags] = None,
- retention: Optional[Retention] = None,
- legal_hold: bool = False,
- ) -> ObjectWriteResult:
- """
- Uploads data from a file to an object in a bucket.
- :param bucket_name: Name of the bucket.
- :param object_name: Object name in the bucket.
- :param file_path: Name of file to upload.
- :param content_type: Content type of the object.
- :param metadata: Any additional metadata to be uploaded along
- with your PUT request.
- :param sse: Server-side encryption.
- :param progress: A progress object
- :param part_size: Multipart part size
- :param num_parallel_uploads: Number of parallel uploads.
- :param tags: :class:`Tags` for the object.
- :param retention: :class:`Retention` configuration object.
- :param legal_hold: Flag to set legal hold for the object.
- :return: :class:`ObjectWriteResult` object.
- Example::
- # Upload data.
- result = client.fput_object(
- "my-bucket", "my-object", "my-filename",
- )
- # Upload data with metadata.
- result = client.fput_object(
- "my-bucket", "my-object", "my-filename",
- metadata={"My-Project": "one"},
- )
- # Upload data with tags, retention and legal-hold.
- date = datetime.utcnow().replace(
- hour=0, minute=0, second=0, microsecond=0,
- ) + timedelta(days=30)
- tags = Tags(for_object=True)
- tags["User"] = "jsmith"
- result = client.fput_object(
- "my-bucket", "my-object", "my-filename",
- tags=tags,
- retention=Retention(GOVERNANCE, date),
- legal_hold=True,
- )
- """
- file_size = os.stat(file_path).st_size
- with open(file_path, "rb") as file_data:
- return self.put_object(
- bucket_name,
- object_name,
- file_data,
- file_size,
- content_type=content_type,
- metadata=cast(Union[DictType, None], metadata),
- sse=sse,
- progress=progress,
- part_size=part_size,
- num_parallel_uploads=num_parallel_uploads,
- tags=tags,
- retention=retention,
- legal_hold=legal_hold,
- )
- def fget_object(
- self,
- bucket_name: str,
- object_name: str,
- file_path: str,
- request_headers: Optional[DictType] = None,
- ssec: Optional[SseCustomerKey] = None,
- version_id: Optional[str] = None,
- extra_query_params: Optional[DictType] = None,
- tmp_file_path: Optional[str] = None,
- progress: Optional[ProgressType] = None,
- ):
- """
- Downloads data of an object to file.
- :param bucket_name: Name of the bucket.
- :param object_name: Object name in the bucket.
- :param file_path: Name of file to download.
- :param request_headers: Any additional headers to be added with GET
- request.
- :param ssec: Server-side encryption customer key.
- :param version_id: Version-ID of the object.
- :param extra_query_params: Extra query parameters for advanced usage.
- :param tmp_file_path: Path to a temporary file.
- :param progress: A progress object
- :return: Object information.
- Example::
- # Download data of an object.
- client.fget_object("my-bucket", "my-object", "my-filename")
- # Download data of an object of version-ID.
- client.fget_object(
- "my-bucket", "my-object", "my-filename",
- version_id="dfbd25b3-abec-4184-a4e8-5a35a5c1174d",
- )
- # Download data of an SSE-C encrypted object.
- client.fget_object(
- "my-bucket", "my-object", "my-filename",
- ssec=SseCustomerKey(b"32byteslongsecretkeymustprovided"),
- )
- """
- check_bucket_name(bucket_name, s3_check=self._base_url.is_aws_host)
- check_object_name(object_name)
- if os.path.isdir(file_path):
- raise ValueError(f"file {file_path} is a directory")
- # Create top level directory if needed.
- makedirs(os.path.dirname(file_path))
- stat = self.stat_object(
- bucket_name,
- object_name,
- ssec,
- version_id=version_id,
- extra_headers=request_headers,
- )
- etag = queryencode(cast(str, stat.etag))
- # Write to a temporary file "file_path.part.minio" before saving.
- tmp_file_path = (
- tmp_file_path or f"{file_path}.{etag}.part.minio"
- )
- response = None
- try:
- response = self.get_object(
- bucket_name,
- object_name,
- request_headers=request_headers,
- ssec=ssec,
- version_id=version_id,
- extra_query_params=extra_query_params,
- )
- if progress:
- # Set progress bar length and object name before upload
- length = int(response.headers.get('content-length', 0))
- progress.set_meta(object_name=object_name, total_length=length)
- with open(tmp_file_path, "wb") as tmp_file:
- for data in response.stream(amt=1024 * 1024):
- size = tmp_file.write(data)
- if progress:
- progress.update(size)
- if os.path.exists(file_path):
- os.remove(file_path) # For windows compatibility.
- os.rename(tmp_file_path, file_path)
- return stat
- finally:
- if response:
- response.close()
- response.release_conn()
- def get_object(
- self,
- bucket_name: str,
- object_name: str,
- offset: int = 0,
- length: int = 0,
- request_headers: Optional[DictType] = None,
- ssec: Optional[SseCustomerKey] = None,
- version_id: Optional[str] = None,
- extra_query_params: Optional[DictType] = None,
- ) -> BaseHTTPResponse:
- """
- Get data of an object. Returned response should be closed after use to
- release network resources. To reuse the connection, it's required to
- call `response.release_conn()` explicitly.
- :param bucket_name: Name of the bucket.
- :param object_name: Object name in the bucket.
- :param offset: Start byte position of object data.
- :param length: Number of bytes of object data from offset.
- :param request_headers: Any additional headers to be added with GET
- request.
- :param ssec: Server-side encryption customer key.
- :param version_id: Version-ID of the object.
- :param extra_query_params: Extra query parameters for advanced usage.
- :return: :class:`urllib3.response.BaseHTTPResponse` object.
- Example::
- # Get data of an object.
- response = None
- try:
- response = client.get_object("my-bucket", "my-object")
- # Read data from response.
- finally:
- if response:
- response.close()
- response.release_conn()
- # Get data of an object of version-ID.
- response = None
- try:
- response = client.get_object(
- "my-bucket", "my-object",
- version_id="dfbd25b3-abec-4184-a4e8-5a35a5c1174d",
- )
- # Read data from response.
- finally:
- if response:
- response.close()
- response.release_conn()
- # Get data of an object from offset and length.
- response = None
- try:
- response = client.get_object(
- "my-bucket", "my-object", offset=512, length=1024,
- )
- # Read data from response.
- finally:
- if response:
- response.close()
- response.release_conn()
- # Get data of an SSE-C encrypted object.
- response = None
- try:
- response = client.get_object(
- "my-bucket", "my-object",
- ssec=SseCustomerKey(b"32byteslongsecretkeymustprovided"),
- )
- # Read data from response.
- finally:
- if response:
- response.close()
- response.release_conn()
- """
- check_bucket_name(bucket_name, s3_check=self._base_url.is_aws_host)
- check_object_name(object_name)
- check_ssec(ssec)
- headers = cast(DictType, ssec.headers() if ssec else {})
- headers.update(request_headers or {})
- if offset or length:
- end = (offset + length - 1) if length else ""
- headers['Range'] = f"bytes={offset}-{end}"
- if version_id:
- extra_query_params = extra_query_params or {}
- extra_query_params["versionId"] = version_id
- return self._execute(
- "GET",
- bucket_name,
- object_name,
- headers=cast(DictType, headers),
- query_params=extra_query_params,
- preload_content=False,
- )
- def prompt_object(
- self,
- bucket_name: str,
- object_name: str,
- prompt: str,
- lambda_arn: Optional[str] = None,
- request_headers: Optional[DictType] = None,
- ssec: Optional[SseCustomerKey] = None,
- version_id: Optional[str] = None,
- **kwargs: Optional[Any],
- ) -> BaseHTTPResponse:
- """
- Prompt an object using natural language.
- :param bucket_name: Name of the bucket.
- :param object_name: Object name in the bucket.
- :param prompt: Prompt the Object to interact with the AI model.
- request.
- :param lambda_arn: Lambda ARN to use for prompt.
- :param request_headers: Any additional headers to be added with POST
- :param ssec: Server-side encryption customer key.
- :param version_id: Version-ID of the object.
- :param kwargs: Extra parameters for advanced usage.
- :return: :class:`urllib3.response.BaseHTTPResponse` object.
- Example::
- # prompt an object.
- response = None
- try:
- response = client.get_object(
- "my-bucket", "my-object",
- "Describe the object for me")
- # Read data from response.
- finally:
- if response:
- response.close()
- response.release_conn()
- """
- check_bucket_name(bucket_name, s3_check=self._base_url.is_aws_host)
- check_object_name(object_name)
- check_ssec(ssec)
- headers = cast(DictType, ssec.headers() if ssec else {})
- headers.update(request_headers or {})
- extra_query_params = {"lambdaArn": lambda_arn or ""}
- if version_id:
- extra_query_params["versionId"] = version_id
- prompt_body = kwargs
- prompt_body["prompt"] = prompt
- body = json.dumps(prompt_body)
- return self._execute(
- "POST",
- bucket_name,
- object_name,
- headers=cast(DictType, headers),
- query_params=cast(DictType, extra_query_params),
- body=body.encode(),
- preload_content=False,
- )
- def copy_object(
- self,
- bucket_name: str,
- object_name: str,
- source: CopySource,
- sse: Optional[Sse] = None,
- metadata: Optional[DictType] = None,
- tags: Optional[Tags] = None,
- retention: Optional[Retention] = None,
- legal_hold: bool = False,
- metadata_directive: Optional[str] = None,
- tagging_directive: Optional[str] = None,
- ) -> ObjectWriteResult:
- """
- Create an object by server-side copying data from another object.
- In this API maximum supported source object size is 5GiB.
- :param bucket_name: Name of the bucket.
- :param object_name: Object name in the bucket.
- :param source: :class:`CopySource` object.
- :param sse: Server-side encryption of destination object.
- :param metadata: Any user-defined metadata to be copied along with
- destination object.
- :param tags: Tags for destination object.
- :param retention: :class:`Retention` configuration object.
- :param legal_hold: Flag to set legal hold for destination object.
- :param metadata_directive: Directive used to handle user metadata for
- destination object.
- :param tagging_directive: Directive used to handle tags for destination
- object.
- :return: :class:`ObjectWriteResult <ObjectWriteResult>` object.
- Example::
- # copy an object from a bucket to another.
- result = client.copy_object(
- "my-bucket",
- "my-object",
- CopySource("my-sourcebucket", "my-sourceobject"),
- )
- print(result.object_name, result.version_id)
- # copy an object with condition.
- result = client.copy_object(
- "my-bucket",
- "my-object",
- CopySource(
- "my-sourcebucket",
- "my-sourceobject",
- modified_since=datetime(2014, 4, 1, tzinfo=timezone.utc),
- ),
- )
- print(result.object_name, result.version_id)
- # copy an object from a bucket with replacing metadata.
- metadata = {"test_meta_key": "test_meta_value"}
- result = client.copy_object(
- "my-bucket",
- "my-object",
- CopySource("my-sourcebucket", "my-sourceobject"),
- metadata=metadata,
- metadata_directive=REPLACE,
- )
- print(result.object_name, result.version_id)
- """
- check_bucket_name(bucket_name, s3_check=self._base_url.is_aws_host)
- check_object_name(object_name)
- if not isinstance(source, CopySource):
- raise ValueError("source must be CopySource type")
- check_sse(sse)
- if tags is not None and not isinstance(tags, Tags):
- raise ValueError("tags must be Tags type")
- if retention is not None and not isinstance(retention, Retention):
- raise ValueError("retention must be Retention type")
- if (
- metadata_directive is not None and
- metadata_directive not in [COPY, REPLACE]
- ):
- raise ValueError(f"metadata directive must be {COPY} or {REPLACE}")
- if (
- tagging_directive is not None and
- tagging_directive not in [COPY, REPLACE]
- ):
- raise ValueError(f"tagging directive must be {COPY} or {REPLACE}")
- size = -1
- if source.offset is None and source.length is None:
- stat = self.stat_object(
- source.bucket_name,
- source.object_name,
- version_id=source.version_id,
- ssec=source.ssec,
- )
- size = cast(int, stat.size)
- if (
- source.offset is not None or
- source.length is not None or
- size > MAX_PART_SIZE
- ):
- if metadata_directive == COPY:
- raise ValueError(
- "COPY metadata directive is not applicable to source "
- "object size greater than 5 GiB",
- )
- if tagging_directive == COPY:
- raise ValueError(
- "COPY tagging directive is not applicable to source "
- "object size greater than 5 GiB"
- )
- return self.compose_object(
- bucket_name, object_name, [ComposeSource.of(source)],
- sse=sse, metadata=metadata, tags=tags, retention=retention,
- legal_hold=legal_hold,
- )
- headers = genheaders(
- metadata,
- sse,
- tags,
- retention,
- legal_hold,
- )
- if metadata_directive:
- headers["x-amz-metadata-directive"] = metadata_directive
- if tagging_directive:
- headers["x-amz-tagging-directive"] = tagging_directive
- headers.update(source.gen_copy_headers())
- response = self._execute(
- "PUT",
- bucket_name,
- object_name=object_name,
- headers=headers,
- )
- etag, last_modified = parse_copy_object(response)
- return ObjectWriteResult(
- bucket_name,
- object_name,
- response.headers.get("x-amz-version-id"),
- etag,
- response.headers,
- last_modified=last_modified,
- )
- def _calc_part_count(self, sources: list[ComposeSource]) -> int:
- """Calculate part count."""
- object_size = 0
- part_count = 0
- i = 0
- for src in sources:
- i += 1
- stat = self.stat_object(
- src.bucket_name,
- src.object_name,
- version_id=src.version_id,
- ssec=src.ssec,
- )
- src.build_headers(cast(int, stat.size), cast(str, stat.etag))
- size = cast(int, stat.size)
- if src.length is not None:
- size = src.length
- elif src.offset is not None:
- size -= src.offset
- if (
- size < MIN_PART_SIZE and
- len(sources) != 1 and
- i != len(sources)
- ):
- raise ValueError(
- f"source {src.bucket_name}/{src.object_name}: size {size} "
- f"must be greater than {MIN_PART_SIZE}"
- )
- object_size += size
- if object_size > MAX_MULTIPART_OBJECT_SIZE:
- raise ValueError(
- f"destination object size must be less than "
- f"{MAX_MULTIPART_OBJECT_SIZE}"
- )
- if size > MAX_PART_SIZE:
- count = int(size / MAX_PART_SIZE)
- last_part_size = size - (count * MAX_PART_SIZE)
- if last_part_size > 0:
- count += 1
- else:
- last_part_size = MAX_PART_SIZE
- if (
- last_part_size < MIN_PART_SIZE and
- len(sources) != 1 and
- i != len(sources)
- ):
- raise ValueError(
- f"source {src.bucket_name}/{src.object_name}: "
- f"for multipart split upload of {size}, "
- f"last part size is less than {MIN_PART_SIZE}"
- )
- part_count += count
- else:
- part_count += 1
- if part_count > MAX_MULTIPART_COUNT:
- raise ValueError(
- f"Compose sources create more than allowed multipart "
- f"count {MAX_MULTIPART_COUNT}"
- )
- return part_count
- def _upload_part_copy(
- self,
- bucket_name: str,
- object_name: str,
- upload_id: str,
- part_number: int,
- headers: DictType,
- ) -> tuple[str, Optional[datetime]]:
- """Execute UploadPartCopy S3 API."""
- response = self._execute(
- "PUT",
- bucket_name,
- object_name,
- headers=headers,
- query_params={
- "partNumber": str(part_number),
- "uploadId": upload_id,
- },
- )
- return parse_copy_object(response)
- def compose_object(
- self,
- bucket_name: str,
- object_name: str,
- sources: list[ComposeSource],
- sse: Optional[Sse] = None,
- metadata: Optional[DictType] = None,
- tags: Optional[Tags] = None,
- retention: Optional[Retention] = None,
- legal_hold: bool = False,
- ) -> ObjectWriteResult:
- """
- Create an object by combining data from different source objects using
- server-side copy.
- :param bucket_name: Name of the bucket.
- :param object_name: Object name in the bucket.
- :param sources: List of :class:`ComposeSource` object.
- :param sse: Server-side encryption of destination object.
- :param metadata: Any user-defined metadata to be copied along with
- destination object.
- :param tags: Tags for destination object.
- :param retention: :class:`Retention` configuration object.
- :param legal_hold: Flag to set legal hold for destination object.
- :return: :class:`ObjectWriteResult <ObjectWriteResult>` object.
- Example::
- sources = [
- ComposeSource("my-job-bucket", "my-object-part-one"),
- ComposeSource("my-job-bucket", "my-object-part-two"),
- ComposeSource("my-job-bucket", "my-object-part-three"),
- ]
- # Create my-bucket/my-object by combining source object
- # list.
- result = client.compose_object("my-bucket", "my-object", sources)
- print(result.object_name, result.version_id)
- # Create my-bucket/my-object with user metadata by combining
- # source object list.
- result = client.compose_object(
- "my-bucket",
- "my-object",
- sources,
- metadata={"test_meta_key": "test_meta_value"},
- )
- print(result.object_name, result.version_id)
- # Create my-bucket/my-object with user metadata and
- # server-side encryption by combining source object list.
- client.compose_object(
- "my-bucket", "my-object", sources, sse=SseS3(),
- )
- print(result.object_name, result.version_id)
- """
- check_bucket_name(bucket_name, s3_check=self._base_url.is_aws_host)
- check_object_name(object_name)
- if not isinstance(sources, (list, tuple)) or not sources:
- raise ValueError("sources must be non-empty list or tuple type")
- i = 0
- for src in sources:
- if not isinstance(src, ComposeSource):
- raise ValueError(f"sources[{i}] must be ComposeSource type")
- i += 1
- check_sse(sse)
- if tags is not None and not isinstance(tags, Tags):
- raise ValueError("tags must be Tags type")
- if retention is not None and not isinstance(retention, Retention):
- raise ValueError("retention must be Retention type")
- part_count = self._calc_part_count(sources)
- if (
- part_count == 1 and
- sources[0].offset is None and
- sources[0].length is None
- ):
- return self.copy_object(
- bucket_name, object_name, CopySource.of(sources[0]),
- sse=sse, metadata=metadata, tags=tags, retention=retention,
- legal_hold=legal_hold,
- metadata_directive=REPLACE if metadata else None,
- tagging_directive=REPLACE if tags else None,
- )
- headers = genheaders(metadata, sse, tags, retention, legal_hold)
- upload_id = self._create_multipart_upload(
- bucket_name, object_name, headers,
- )
- ssec_headers = sse.headers() if isinstance(sse, SseCustomerKey) else {}
- try:
- part_number = 0
- total_parts = []
- for src in sources:
- size = cast(int, src.object_size)
- if src.length is not None:
- size = src.length
- elif src.offset is not None:
- size -= src.offset
- offset = src.offset or 0
- headers = cast(DictType, src.headers)
- headers.update(ssec_headers)
- if size <= MAX_PART_SIZE:
- part_number += 1
- if src.length is not None:
- headers["x-amz-copy-source-range"] = (
- f"bytes={offset}-{offset + src.length - 1}"
- )
- elif src.offset is not None:
- headers["x-amz-copy-source-range"] = (
- f"bytes={offset}-{offset + size - 1}"
- )
- etag, _ = self._upload_part_copy(
- bucket_name,
- object_name,
- upload_id,
- part_number,
- headers,
- )
- total_parts.append(Part(part_number, etag))
- continue
- while size > 0:
- part_number += 1
- length = size if size < MAX_PART_SIZE else MAX_PART_SIZE
- end_bytes = offset + length - 1
- headers_copy = headers.copy()
- headers_copy["x-amz-copy-source-range"] = (
- f"bytes={offset}-{end_bytes}"
- )
- etag, _ = self._upload_part_copy(
- bucket_name,
- object_name,
- upload_id,
- part_number,
- headers_copy,
- )
- total_parts.append(Part(part_number, etag))
- offset += length
- size -= length
- result = self._complete_multipart_upload(
- bucket_name, object_name, upload_id, total_parts,
- sse if isinstance(sse, SseCustomerKey) else None,
- )
- return ObjectWriteResult(
- cast(str, result.bucket_name),
- cast(str, result.object_name),
- result.version_id,
- result.etag,
- result.http_headers,
- location=result.location,
- )
- except Exception as exc:
- if upload_id:
- self._abort_multipart_upload(
- bucket_name, object_name, upload_id,
- )
- raise exc
- def _abort_multipart_upload(
- self,
- bucket_name: str,
- object_name: str,
- upload_id: str,
- ):
- """Execute AbortMultipartUpload S3 API."""
- self._execute(
- "DELETE",
- bucket_name,
- object_name,
- query_params={'uploadId': upload_id},
- )
- def _complete_multipart_upload(
- self,
- bucket_name: str,
- object_name: str,
- upload_id: str,
- parts: list[Part],
- ssec: Optional[SseCustomerKey] = None,
- ) -> CompleteMultipartUploadResult:
- """Execute CompleteMultipartUpload S3 API."""
- element = Element("CompleteMultipartUpload")
- for part in parts:
- tag = SubElement(element, "Part")
- SubElement(tag, "PartNumber", str(part.part_number))
- SubElement(tag, "ETag", '"' + part.etag + '"')
- body = getbytes(element)
- headers: DictType = {
- "Content-Type": 'application/xml',
- "Content-MD5": cast(str, md5sum_hash(body)),
- }
- if ssec:
- headers.update(ssec.headers())
- response = self._execute(
- "POST",
- bucket_name,
- object_name,
- body=body,
- headers=headers,
- query_params={'uploadId': upload_id},
- )
- return CompleteMultipartUploadResult(response)
- def _create_multipart_upload(
- self,
- bucket_name: str,
- object_name: str,
- headers: DictType,
- ) -> str:
- """Execute CreateMultipartUpload S3 API."""
- if not headers.get("Content-Type"):
- headers["Content-Type"] = "application/octet-stream"
- response = self._execute(
- "POST",
- bucket_name,
- object_name,
- headers=headers,
- query_params={"uploads": ""},
- )
- element = ET.fromstring(response.data.decode())
- return cast(str, findtext(element, "UploadId", True))
- def _put_object(
- self,
- bucket_name: str,
- object_name: str,
- data: bytes,
- headers: Optional[DictType] = None,
- query_params: Optional[DictType] = None,
- ) -> ObjectWriteResult:
- """Execute PutObject S3 API."""
- response = self._execute(
- "PUT",
- bucket_name,
- object_name,
- body=data,
- headers=headers,
- query_params=query_params,
- no_body_trace=True,
- )
- return ObjectWriteResult(
- bucket_name,
- object_name,
- response.headers.get("x-amz-version-id"),
- response.headers.get("etag", "").replace('"', ""),
- response.headers,
- )
- def _upload_part(
- self,
- bucket_name: str,
- object_name: str,
- data: bytes,
- headers: Optional[DictType],
- upload_id: str,
- part_number: int,
- ) -> str:
- """Execute UploadPart S3 API."""
- result = self._put_object(
- bucket_name,
- object_name,
- data,
- headers,
- query_params={
- "partNumber": str(part_number),
- "uploadId": upload_id,
- },
- )
- return cast(str, result.etag)
- def _upload_part_task(self, args):
- """Upload_part task for ThreadPool."""
- return args[5], self._upload_part(*args)
- def put_object(
- self,
- bucket_name: str,
- object_name: str,
- data: BinaryIO,
- length: int,
- content_type: str = "application/octet-stream",
- metadata: Optional[DictType] = None,
- sse: Optional[Sse] = None,
- progress: Optional[ProgressType] = None,
- part_size: int = 0,
- num_parallel_uploads: int = 3,
- tags: Optional[Tags] = None,
- retention: Optional[Retention] = None,
- legal_hold: bool = False,
- write_offset: Optional[int] = None,
- ) -> ObjectWriteResult:
- """
- Uploads data from a stream to an object in a bucket.
- :param bucket_name: Name of the bucket.
- :param object_name: Object name in the bucket.
- :param data: An object having callable read() returning bytes object.
- :param length: Data size; -1 for unknown size and set valid part_size.
- :param content_type: Content type of the object.
- :param metadata: Any additional metadata to be uploaded along
- with your PUT request.
- :param sse: Server-side encryption.
- :param progress: A progress object;
- :param part_size: Multipart part size.
- :param num_parallel_uploads: Number of parallel uploads.
- :param tags: :class:`Tags` for the object.
- :param retention: :class:`Retention` configuration object.
- :param legal_hold: Flag to set legal hold for the object.
- :param write_offset: Offset byte for appending data to existing object.
- :return: :class:`ObjectWriteResult` object.
- Example::
- # Upload data.
- result = client.put_object(
- "my-bucket", "my-object", io.BytesIO(b"hello"), 5,
- )
- # Upload data with metadata.
- result = client.put_object(
- "my-bucket", "my-object", io.BytesIO(b"hello"), 5,
- metadata={"My-Project": "one"},
- )
- # Upload data with tags, retention and legal-hold.
- date = datetime.utcnow().replace(
- hour=0, minute=0, second=0, microsecond=0,
- ) + timedelta(days=30)
- tags = Tags(for_object=True)
- tags["User"] = "jsmith"
- result = client.put_object(
- "my-bucket", "my-object", io.BytesIO(b"hello"), 5,
- tags=tags,
- retention=Retention(GOVERNANCE, date),
- legal_hold=True,
- )
- """
- check_bucket_name(bucket_name, s3_check=self._base_url.is_aws_host)
- check_object_name(object_name)
- check_sse(sse)
- if tags is not None and not isinstance(tags, Tags):
- raise ValueError("tags must be Tags type")
- if retention is not None and not isinstance(retention, Retention):
- raise ValueError("retention must be Retention type")
- if not callable(getattr(data, "read")):
- raise ValueError("input data must have callable read()")
- if write_offset is not None:
- if write_offset < 0:
- raise ValueError("write offset should not be negative")
- if length < 0:
- raise ValueError("length must be provided for write offset")
- part_size = length if length > MIN_PART_SIZE else MIN_PART_SIZE
- part_size, part_count = get_part_info(length, part_size)
- if progress:
- # Set progress bar length and object name before upload
- progress.set_meta(object_name=object_name, total_length=length)
- headers = genheaders(metadata, sse, tags, retention, legal_hold)
- headers["Content-Type"] = content_type or "application/octet-stream"
- if write_offset:
- headers["x-amz-write-offset-bytes"] = str(write_offset)
- object_size = length
- uploaded_size = 0
- part_number = 0
- one_byte = b""
- stop = False
- upload_id = None
- parts: list[Part] = []
- pool: Optional[ThreadPool] = None
- try:
- while not stop:
- part_number += 1
- if part_count > 0:
- if part_number == part_count:
- part_size = object_size - uploaded_size
- stop = True
- part_data = read_part_data(
- data, part_size, progress=progress,
- )
- if len(part_data) != part_size:
- raise IOError(
- f"stream having not enough data;"
- f"expected: {part_size}, "
- f"got: {len(part_data)} bytes"
- )
- else:
- part_data = read_part_data(
- data, part_size + 1, one_byte, progress=progress,
- )
- # If part_data_size is less or equal to part_size,
- # then we have reached last part.
- if len(part_data) <= part_size:
- part_count = part_number
- stop = True
- else:
- one_byte = part_data[-1:]
- part_data = part_data[:-1]
- uploaded_size += len(part_data)
- if part_count == 1:
- return self._put_object(
- bucket_name, object_name, part_data, headers,
- )
- if not upload_id:
- upload_id = self._create_multipart_upload(
- bucket_name, object_name, headers,
- )
- if num_parallel_uploads and num_parallel_uploads > 1:
- pool = ThreadPool(num_parallel_uploads)
- pool.start_parallel()
- args = (
- bucket_name,
- object_name,
- part_data,
- (
- cast(DictType, sse.headers())
- if isinstance(sse, SseCustomerKey) else None
- ),
- upload_id,
- part_number,
- )
- if num_parallel_uploads > 1:
- cast(ThreadPool, pool).add_task(
- self._upload_part_task, args,
- )
- else:
- etag = self._upload_part(*args)
- parts.append(Part(part_number, etag))
- if pool:
- result = pool.result()
- parts = [Part(0, "")] * part_count
- while not result.empty():
- part_number, etag = result.get()
- parts[part_number - 1] = Part(part_number, etag)
- upload_result = self._complete_multipart_upload(
- bucket_name, object_name, cast(str, upload_id), parts,
- sse if isinstance(sse, SseCustomerKey) else None,
- )
- return ObjectWriteResult(
- cast(str, upload_result.bucket_name),
- cast(str, upload_result.object_name),
- upload_result.version_id,
- upload_result.etag,
- upload_result.http_headers,
- location=upload_result.location,
- )
- except Exception as exc:
- if upload_id:
- self._abort_multipart_upload(
- bucket_name, object_name, upload_id,
- )
- raise exc
- def append_object(
- self,
- bucket_name: str,
- object_name: str,
- data: BinaryIO,
- length: int,
- chunk_size: Optional[int] = None,
- progress: Optional[ProgressType] = None,
- extra_headers: Optional[DictType] = None,
- ) -> ObjectWriteResult:
- """
- Appends from a stream to existing object in a bucket.
- :param bucket_name: Name of the bucket.
- :param object_name: Object name in the bucket.
- :param data: An object having callable read() returning bytes object.
- :param length: Data size; -1 for unknown size.
- :param chunk_size: Chunk size to optimize uploads.
- :return: :class:`ObjectWriteResult` object.
- Example::
- # Append data.
- result = client.append_object(
- "my-bucket", "my-object", io.BytesIO(b"world"), 5,
- )
- print(f"appended {result.object_name} object; etag: {result.etag}")
- # Append data in chunks.
- data = urlopen(
- "https://www.kernel.org/pub/linux/kernel/v6.x/"
- "linux-6.13.12.tar.xz",
- )
- result = client.append_object(
- "my-bucket", "my-object", data, 148611164, 5*1024*1024,
- )
- print(f"appended {result.object_name} object; etag: {result.etag}")
- # Append unknown sized data.
- data = urlopen(
- "https://www.kernel.org/pub/linux/kernel/v6.x/"
- "linux-6.14.3.tar.xz",
- )
- result = client.append_object(
- "my-bucket", "my-object", data, 149426584, 5*1024*1024,
- )
- print(f"appended {result.object_name} object; etag: {result.etag}")
- """
- if length == 0:
- raise ValueError("length should not be zero")
- if chunk_size is not None:
- if chunk_size < MIN_PART_SIZE:
- raise ValueError("chunk size must be minimum of 5 MiB")
- if chunk_size > MAX_PART_SIZE:
- raise ValueError("chunk size must be less than 5 GiB")
- else:
- chunk_size = length if length > MIN_PART_SIZE else MIN_PART_SIZE
- chunk_count = -1
- if length > 0:
- chunk_count = int(length / chunk_size)
- if (chunk_count * chunk_size) < length:
- chunk_count += 1
- chunk_count = chunk_count or 1
- object_size = length
- uploaded_size = 0
- chunk_number = 0
- one_byte = b""
- stop = False
- stat = self.stat_object(bucket_name, object_name)
- write_offset = cast(int, stat.size)
- while not stop:
- chunk_number += 1
- if chunk_count > 0:
- if chunk_number == chunk_count:
- chunk_size = object_size - uploaded_size
- stop = True
- chunk_data = read_part_data(
- data, chunk_size, progress=progress,
- )
- if len(chunk_data) != chunk_size:
- raise IOError(
- f"stream having not enough data;"
- f"expected: {chunk_size}, "
- f"got: {len(chunk_data)} bytes"
- )
- else:
- chunk_data = read_part_data(
- data, chunk_size + 1, one_byte, progress=progress,
- )
- # If chunk_data_size is less or equal to chunk_size,
- # then we have reached last chunk.
- if len(chunk_data) <= chunk_size:
- chunk_count = chunk_number
- stop = True
- else:
- one_byte = chunk_data[-1:]
- chunk_data = chunk_data[:-1]
- uploaded_size += len(chunk_data)
- headers = extra_headers or {}
- headers["x-amz-write-offset-bytes"] = str(write_offset)
- upload_result = self._put_object(
- bucket_name, object_name, chunk_data, headers=headers,
- )
- write_offset += len(chunk_data)
- return ObjectWriteResult(
- cast(str, upload_result.bucket_name),
- cast(str, upload_result.object_name),
- upload_result.version_id,
- upload_result.etag,
- upload_result.http_headers,
- location=upload_result.location,
- )
- def list_objects(
- self,
- bucket_name: str,
- prefix: Optional[str] = None,
- recursive: bool = False,
- start_after: Optional[str] = None,
- include_user_meta: bool = False,
- include_version: bool = False,
- use_api_v1: bool = False,
- use_url_encoding_type: bool = True,
- fetch_owner: bool = False,
- extra_headers: Optional[DictType] = None,
- extra_query_params: Optional[DictType] = None,
- ):
- """
- Lists object information of a bucket.
- :param bucket_name: Name of the bucket.
- :param prefix: Object name starts with prefix.
- :param recursive: List recursively than directory structure emulation.
- :param start_after: List objects after this key name.
- :param include_user_meta: MinIO specific flag to control to include
- user metadata.
- :param include_version: Flag to control whether include object
- versions.
- :param use_api_v1: Flag to control to use ListObjectV1 S3 API or not.
- :param use_url_encoding_type: Flag to control whether URL encoding type
- to be used or not.
- :param extra_headers: Extra HTTP headers for advanced usage.
- :param extra_query_params: Extra query parameters for advanced usage.
- :return: Iterator of :class:`Object <Object>`.
- Example::
- # List objects information.
- objects = client.list_objects("my-bucket")
- for obj in objects:
- print(obj)
- # List objects information whose names starts with "my/prefix/".
- objects = client.list_objects("my-bucket", prefix="my/prefix/")
- for obj in objects:
- print(obj)
- # List objects information recursively.
- objects = client.list_objects("my-bucket", recursive=True)
- for obj in objects:
- print(obj)
- # List objects information recursively whose names starts with
- # "my/prefix/".
- objects = client.list_objects(
- "my-bucket", prefix="my/prefix/", recursive=True,
- )
- for obj in objects:
- print(obj)
- # List objects information recursively after object name
- # "my/prefix/world/1".
- objects = client.list_objects(
- "my-bucket", recursive=True, start_after="my/prefix/world/1",
- )
- for obj in objects:
- print(obj)
- """
- return self._list_objects(
- bucket_name,
- delimiter=None if recursive else "/",
- include_user_meta=include_user_meta,
- prefix=prefix,
- start_after=start_after,
- use_api_v1=use_api_v1,
- include_version=include_version,
- encoding_type="url" if use_url_encoding_type else None,
- fetch_owner=fetch_owner,
- extra_headers=extra_headers,
- extra_query_params=extra_query_params,
- )
- def stat_object(
- self,
- bucket_name: str,
- object_name: str,
- ssec: Optional[SseCustomerKey] = None,
- version_id: Optional[str] = None,
- extra_headers: Optional[DictType] = None,
- extra_query_params: Optional[DictType] = None,
- ) -> Object:
- """
- Get object information and metadata of an object.
- :param bucket_name: Name of the bucket.
- :param object_name: Object name in the bucket.
- :param ssec: Server-side encryption customer key.
- :param version_id: Version ID of the object.
- :param extra_headers: Extra HTTP headers for advanced usage.
- :param extra_query_params: Extra query parameters for advanced usage.
- :return: :class:`Object <Object>`.
- Example::
- # Get object information.
- result = client.stat_object("my-bucket", "my-object")
- # Get object information of version-ID.
- result = client.stat_object(
- "my-bucket", "my-object",
- version_id="dfbd25b3-abec-4184-a4e8-5a35a5c1174d",
- )
- # Get SSE-C encrypted object information.
- result = client.stat_object(
- "my-bucket", "my-object",
- ssec=SseCustomerKey(b"32byteslongsecretkeymustprovided"),
- )
- """
- check_bucket_name(bucket_name, s3_check=self._base_url.is_aws_host)
- check_object_name(object_name)
- check_ssec(ssec)
- headers = cast(DictType, ssec.headers() if ssec else {})
- if extra_headers:
- headers.update(extra_headers)
- query_params = extra_query_params or {}
- query_params.update({"versionId": version_id} if version_id else {})
- response = self._execute(
- "HEAD",
- bucket_name,
- object_name,
- headers=headers,
- query_params=query_params,
- )
- value = response.headers.get("last-modified")
- if value is not None:
- last_modified = time.from_http_header(value)
- else:
- last_modified = None
- return Object(
- bucket_name,
- object_name,
- last_modified=last_modified,
- etag=response.headers.get("etag", "").replace('"', ""),
- size=int(response.headers.get("content-length", "0")),
- content_type=response.headers.get("content-type"),
- metadata=response.headers,
- version_id=response.headers.get("x-amz-version-id"),
- )
- def remove_object(
- self,
- bucket_name: str,
- object_name: str,
- version_id: Optional[str] = None
- ):
- """
- Remove an object.
- :param bucket_name: Name of the bucket.
- :param object_name: Object name in the bucket.
- :param version_id: Version ID of the object.
- Example::
- # Remove object.
- client.remove_object("my-bucket", "my-object")
- # Remove version of an object.
- client.remove_object(
- "my-bucket", "my-object",
- version_id="dfbd25b3-abec-4184-a4e8-5a35a5c1174d",
- )
- """
- check_bucket_name(bucket_name, s3_check=self._base_url.is_aws_host)
- check_object_name(object_name)
- self._execute(
- "DELETE",
- bucket_name,
- object_name,
- query_params={"versionId": version_id} if version_id else None,
- )
- def _delete_objects(
- self,
- bucket_name: str,
- delete_object_list: list[DeleteObject],
- quiet: bool = False,
- bypass_governance_mode: bool = False,
- ) -> DeleteResult:
- """
- Delete multiple objects.
- :param bucket_name: Name of the bucket.
- :param delete_object_list: List of maximum 1000
- :class:`DeleteObject <DeleteObject>` object.
- :param quiet: quiet flag.
- :param bypass_governance_mode: Bypass Governance retention mode.
- :return: :class:`DeleteResult <DeleteResult>` object.
- """
- body = marshal(DeleteRequest(delete_object_list, quiet=quiet))
- headers: DictType = {
- "Content-MD5": cast(str, md5sum_hash(body)),
- }
- if bypass_governance_mode:
- headers["x-amz-bypass-governance-retention"] = "true"
- response = self._execute(
- "POST",
- bucket_name,
- body=body,
- headers=headers,
- query_params={"delete": ""},
- )
- element = ET.fromstring(response.data.decode())
- return (
- DeleteResult([], [DeleteError.fromxml(element)])
- if element.tag.endswith("Error")
- else unmarshal(DeleteResult, response.data.decode())
- )
- def remove_objects(
- self,
- bucket_name: str,
- delete_object_list: Iterable[DeleteObject],
- bypass_governance_mode: bool = False,
- ) -> Iterator[DeleteError]:
- """
- Remove multiple objects.
- :param bucket_name: Name of the bucket.
- :param delete_object_list: An iterable containing
- :class:`DeleteObject <DeleteObject>` object.
- :param bypass_governance_mode: Bypass Governance retention mode.
- :return: An iterator containing :class:`DeleteError <DeleteError>`
- object.
- Example::
- # Remove list of objects.
- errors = client.remove_objects(
- "my-bucket",
- [
- DeleteObject("my-object1"),
- DeleteObject("my-object2"),
- DeleteObject(
- "my-object3", "13f88b18-8dcd-4c83-88f2-8631fdb6250c",
- ),
- ],
- )
- for error in errors:
- print("error occurred when deleting object", error)
- # Remove a prefix recursively.
- delete_object_list = list(
- map(
- lambda x: DeleteObject(x.object_name),
- client.list_objects(
- "my-bucket",
- "my/prefix/",
- recursive=True,
- ),
- )
- )
- errors = client.remove_objects("my-bucket", delete_object_list)
- for error in errors:
- print("error occurred when deleting object", error)
- """
- check_bucket_name(bucket_name, s3_check=self._base_url.is_aws_host)
- # turn list like objects into an iterator.
- delete_object_list = itertools.chain(delete_object_list)
- while True:
- # get 1000 entries or whatever available.
- objects = [
- delete_object for _, delete_object in zip(
- range(1000), delete_object_list,
- )
- ]
- if not objects:
- break
- result = self._delete_objects(
- bucket_name,
- objects,
- quiet=True,
- bypass_governance_mode=bypass_governance_mode,
- )
- for error in result.error_list:
- # AWS S3 returns "NoSuchVersion" error when
- # version doesn't exist ignore this error
- # yield all errors otherwise
- if error.code != "NoSuchVersion":
- yield error
- def get_presigned_url(
- self,
- method: str,
- bucket_name: str,
- object_name: str,
- expires: timedelta = timedelta(days=7),
- response_headers: Optional[DictType] = None,
- request_date: Optional[datetime] = None,
- version_id: Optional[str] = None,
- extra_query_params: Optional[DictType] = None,
- ) -> str:
- """
- Get presigned URL of an object for HTTP method, expiry time and custom
- request parameters.
- :param method: HTTP method.
- :param bucket_name: Name of the bucket.
- :param object_name: Object name in the bucket.
- :param expires: Expiry in seconds; defaults to 7 days.
- :param response_headers: Optional response_headers argument to
- specify response fields like date, size,
- type of file, data about server, etc.
- :param request_date: Optional request_date argument to
- specify a different request date. Default is
- current date.
- :param version_id: Version ID of the object.
- :param extra_query_params: Extra query parameters for advanced usage.
- :return: URL string.
- Example::
- # Get presigned URL string to delete 'my-object' in
- # 'my-bucket' with one day expiry.
- url = client.get_presigned_url(
- "DELETE",
- "my-bucket",
- "my-object",
- expires=timedelta(days=1),
- )
- print(url)
- """
- check_bucket_name(bucket_name, s3_check=self._base_url.is_aws_host)
- check_object_name(object_name)
- if expires.total_seconds() < 1 or expires.total_seconds() > 604800:
- raise ValueError("expires must be between 1 second to 7 days")
- region = self._get_region(bucket_name)
- query_params = extra_query_params or {}
- query_params.update({"versionId": version_id} if version_id else {})
- query_params.update(response_headers or {})
- creds = self._provider.retrieve() if self._provider else None
- if creds and creds.session_token:
- query_params["X-Amz-Security-Token"] = creds.session_token
- url = self._base_url.build(
- method=method,
- region=region,
- bucket_name=bucket_name,
- object_name=object_name,
- query_params=query_params,
- )
- if creds:
- url = presign_v4(
- method=method,
- url=url,
- region=region,
- credentials=creds,
- date=request_date or time.utcnow(),
- expires=int(expires.total_seconds()),
- )
- return urlunsplit(url)
- def presigned_get_object(
- self,
- bucket_name: str,
- object_name: str,
- expires: timedelta = timedelta(days=7),
- response_headers: Optional[DictType] = None,
- request_date: Optional[datetime] = None,
- version_id: Optional[str] = None,
- extra_query_params: Optional[DictType] = None,
- ) -> str:
- """
- Get presigned URL of an object to download its data with expiry time
- and custom request parameters.
- :param bucket_name: Name of the bucket.
- :param object_name: Object name in the bucket.
- :param expires: Expiry in seconds; defaults to 7 days.
- :param response_headers: Optional response_headers argument to
- specify response fields like date, size,
- type of file, data about server, etc.
- :param request_date: Optional request_date argument to
- specify a different request date. Default is
- current date.
- :param version_id: Version ID of the object.
- :param extra_query_params: Extra query parameters for advanced usage.
- :return: URL string.
- Example::
- # Get presigned URL string to download 'my-object' in
- # 'my-bucket' with default expiry (i.e. 7 days).
- url = client.presigned_get_object("my-bucket", "my-object")
- print(url)
- # Get presigned URL string to download 'my-object' in
- # 'my-bucket' with two hours expiry.
- url = client.presigned_get_object(
- "my-bucket", "my-object", expires=timedelta(hours=2),
- )
- print(url)
- """
- return self.get_presigned_url(
- "GET",
- bucket_name,
- object_name,
- expires,
- response_headers=response_headers,
- request_date=request_date,
- version_id=version_id,
- extra_query_params=extra_query_params,
- )
- def presigned_put_object(
- self,
- bucket_name: str,
- object_name: str,
- expires: timedelta = timedelta(days=7),
- ) -> str:
- """
- Get presigned URL of an object to upload data with expiry time and
- custom request parameters.
- :param bucket_name: Name of the bucket.
- :param object_name: Object name in the bucket.
- :param expires: Expiry in seconds; defaults to 7 days.
- :return: URL string.
- Example::
- # Get presigned URL string to upload data to 'my-object' in
- # 'my-bucket' with default expiry (i.e. 7 days).
- url = client.presigned_put_object("my-bucket", "my-object")
- print(url)
- # Get presigned URL string to upload data to 'my-object' in
- # 'my-bucket' with two hours expiry.
- url = client.presigned_put_object(
- "my-bucket", "my-object", expires=timedelta(hours=2),
- )
- print(url)
- """
- return self.get_presigned_url(
- "PUT", bucket_name, object_name, expires,
- )
- def presigned_post_policy(self, policy: PostPolicy) -> dict[str, str]:
- """
- Get form-data of PostPolicy of an object to upload its data using POST
- method.
- :param policy: :class:`PostPolicy <PostPolicy>`.
- :return: :dict: contains form-data.
- Example::
- policy = PostPolicy(
- "my-bucket", datetime.utcnow() + timedelta(days=10),
- )
- policy.add_starts_with_condition("key", "my/object/prefix/")
- policy.add_content_length_range_condition(
- 1*1024*1024, 10*1024*1024,
- )
- form_data = client.presigned_post_policy(policy)
- """
- if not isinstance(policy, PostPolicy):
- raise ValueError("policy must be PostPolicy type")
- if not self._provider:
- raise ValueError(
- "anonymous access does not require presigned post form-data",
- )
- check_bucket_name(
- policy.bucket_name, s3_check=self._base_url.is_aws_host)
- return policy.form_data(
- self._provider.retrieve(),
- self._get_region(policy.bucket_name),
- )
- def delete_bucket_replication(self, bucket_name: str):
- """
- Delete replication configuration of a bucket.
- :param bucket_name: Name of the bucket.
- Example::
- client.delete_bucket_replication("my-bucket")
- """
- check_bucket_name(bucket_name, s3_check=self._base_url.is_aws_host)
- self._execute("DELETE", bucket_name, query_params={"replication": ""})
- def get_bucket_replication(
- self,
- bucket_name: str,
- ) -> Optional[ReplicationConfig]:
- """
- Get bucket replication configuration of a bucket.
- :param bucket_name: Name of the bucket.
- :return: :class:`ReplicationConfig <ReplicationConfig>` object.
- Example::
- config = client.get_bucket_replication("my-bucket")
- """
- check_bucket_name(bucket_name, s3_check=self._base_url.is_aws_host)
- try:
- response = self._execute(
- "GET", bucket_name, query_params={"replication": ""},
- )
- return unmarshal(ReplicationConfig, response.data.decode())
- except S3Error as exc:
- if exc.code != "ReplicationConfigurationNotFoundError":
- raise
- return None
- def set_bucket_replication(
- self,
- bucket_name: str,
- config: ReplicationConfig,
- ):
- """
- Set bucket replication configuration to a bucket.
- :param bucket_name: Name of the bucket.
- :param config: :class:`ReplicationConfig <ReplicationConfig>` object.
- Example::
- config = ReplicationConfig(
- "REPLACE-WITH-ACTUAL-ROLE",
- [
- Rule(
- Destination(
- "REPLACE-WITH-ACTUAL-DESTINATION-BUCKET-ARN",
- ),
- ENABLED,
- delete_marker_replication=DeleteMarkerReplication(
- DISABLED,
- ),
- rule_filter=Filter(
- AndOperator(
- "TaxDocs",
- {"key1": "value1", "key2": "value2"},
- ),
- ),
- rule_id="rule1",
- priority=1,
- ),
- ],
- )
- client.set_bucket_replication("my-bucket", config)
- """
- check_bucket_name(bucket_name, s3_check=self._base_url.is_aws_host)
- if not isinstance(config, ReplicationConfig):
- raise ValueError("config must be ReplicationConfig type")
- body = marshal(config)
- self._execute(
- "PUT",
- bucket_name,
- body=body,
- headers={"Content-MD5": cast(str, md5sum_hash(body))},
- query_params={"replication": ""},
- )
- def delete_bucket_lifecycle(self, bucket_name: str):
- """
- Delete notification configuration of a bucket.
- :param bucket_name: Name of the bucket.
- Example::
- client.delete_bucket_lifecycle("my-bucket")
- """
- check_bucket_name(bucket_name, s3_check=self._base_url.is_aws_host)
- self._execute("DELETE", bucket_name, query_params={"lifecycle": ""})
- def get_bucket_lifecycle(
- self,
- bucket_name: str,
- ) -> Optional[LifecycleConfig]:
- """
- Get bucket lifecycle configuration of a bucket.
- :param bucket_name: Name of the bucket.
- :return: :class:`LifecycleConfig <LifecycleConfig>` object.
- Example::
- config = client.get_bucket_lifecycle("my-bucket")
- """
- check_bucket_name(bucket_name, s3_check=self._base_url.is_aws_host)
- try:
- response = self._execute(
- "GET", bucket_name, query_params={"lifecycle": ""},
- )
- return unmarshal(LifecycleConfig, response.data.decode())
- except S3Error as exc:
- if exc.code != "NoSuchLifecycleConfiguration":
- raise
- return None
- def set_bucket_lifecycle(
- self,
- bucket_name: str,
- config: LifecycleConfig,
- ):
- """
- Set bucket lifecycle configuration to a bucket.
- :param bucket_name: Name of the bucket.
- :param config: :class:`LifecycleConfig <LifecycleConfig>` object.
- Example::
- config = LifecycleConfig(
- [
- Rule(
- ENABLED,
- rule_filter=Filter(prefix="documents/"),
- rule_id="rule1",
- transition=Transition(
- days=30, storage_class="GLACIER",
- ),
- ),
- Rule(
- ENABLED,
- rule_filter=Filter(prefix="logs/"),
- rule_id="rule2",
- expiration=Expiration(days=365),
- ),
- ],
- )
- client.set_bucket_lifecycle("my-bucket", config)
- """
- check_bucket_name(bucket_name, s3_check=self._base_url.is_aws_host)
- if not isinstance(config, LifecycleConfig):
- raise ValueError("config must be LifecycleConfig type")
- body = marshal(config)
- self._execute(
- "PUT",
- bucket_name,
- body=body,
- headers={"Content-MD5": cast(str, md5sum_hash(body))},
- query_params={"lifecycle": ""},
- )
- def delete_bucket_tags(self, bucket_name: str):
- """
- Delete tags configuration of a bucket.
- :param bucket_name: Name of the bucket.
- Example::
- client.delete_bucket_tags("my-bucket")
- """
- check_bucket_name(bucket_name, s3_check=self._base_url.is_aws_host)
- self._execute("DELETE", bucket_name, query_params={"tagging": ""})
- def get_bucket_tags(self, bucket_name: str) -> Optional[Tags]:
- """
- Get tags configuration of a bucket.
- :param bucket_name: Name of the bucket.
- :return: :class:`Tags <Tags>` object.
- Example::
- tags = client.get_bucket_tags("my-bucket")
- """
- check_bucket_name(bucket_name, s3_check=self._base_url.is_aws_host)
- try:
- response = self._execute(
- "GET", bucket_name, query_params={"tagging": ""},
- )
- tagging = unmarshal(Tagging, response.data.decode())
- return tagging.tags
- except S3Error as exc:
- if exc.code != "NoSuchTagSet":
- raise
- return None
- def set_bucket_tags(self, bucket_name: str, tags: Tags):
- """
- Set tags configuration to a bucket.
- :param bucket_name: Name of the bucket.
- :param tags: :class:`Tags <Tags>` object.
- Example::
- tags = Tags.new_bucket_tags()
- tags["Project"] = "Project One"
- tags["User"] = "jsmith"
- client.set_bucket_tags("my-bucket", tags)
- """
- check_bucket_name(bucket_name, s3_check=self._base_url.is_aws_host)
- if not isinstance(tags, Tags):
- raise ValueError("tags must be Tags type")
- body = marshal(Tagging(tags))
- self._execute(
- "PUT",
- bucket_name,
- body=body,
- headers={"Content-MD5": cast(str, md5sum_hash(body))},
- query_params={"tagging": ""},
- )
- def delete_object_tags(
- self,
- bucket_name: str,
- object_name: str,
- version_id: Optional[str] = None,
- ):
- """
- Delete tags configuration of an object.
- :param bucket_name: Name of the bucket.
- :param object_name: Object name in the bucket.
- :param version_id: Version ID of the Object.
- Example::
- client.delete_object_tags("my-bucket", "my-object")
- """
- check_bucket_name(bucket_name, s3_check=self._base_url.is_aws_host)
- check_object_name(object_name)
- query_params = {"versionId": version_id} if version_id else {}
- query_params["tagging"] = ""
- self._execute(
- "DELETE",
- bucket_name,
- object_name=object_name,
- query_params=cast(DictType, query_params),
- )
- def get_object_tags(
- self,
- bucket_name: str,
- object_name: str,
- version_id: Optional[str] = None,
- ) -> Optional[Tags]:
- """
- Get tags configuration of a object.
- :param bucket_name: Name of the bucket.
- :param object_name: Object name in the bucket.
- :param version_id: Version ID of the Object.
- :return: :class:`Tags <Tags>` object.
- Example::
- tags = client.get_object_tags("my-bucket", "my-object")
- """
- check_bucket_name(bucket_name, s3_check=self._base_url.is_aws_host)
- check_object_name(object_name)
- query_params = {"versionId": version_id} if version_id else {}
- query_params["tagging"] = ""
- try:
- response = self._execute(
- "GET",
- bucket_name,
- object_name=object_name,
- query_params=cast(DictType, query_params),
- )
- tagging = unmarshal(Tagging, response.data.decode())
- return tagging.tags
- except S3Error as exc:
- if exc.code != "NoSuchTagSet":
- raise
- return None
- def set_object_tags(
- self,
- bucket_name: str,
- object_name: str,
- tags: Tags,
- version_id: Optional[str] = None,
- ):
- """
- Set tags configuration to an object.
- :param bucket_name: Name of the bucket.
- :param object_name: Object name in the bucket.
- :param version_id: Version ID of the Object.
- :param tags: :class:`Tags <Tags>` object.
- Example::
- tags = Tags.new_object_tags()
- tags["Project"] = "Project One"
- tags["User"] = "jsmith"
- client.set_object_tags("my-bucket", "my-object", tags)
- """
- check_bucket_name(bucket_name, s3_check=self._base_url.is_aws_host)
- check_object_name(object_name)
- if not isinstance(tags, Tags):
- raise ValueError("tags must be Tags type")
- body = marshal(Tagging(tags))
- query_params = {"versionId": version_id} if version_id else {}
- query_params["tagging"] = ""
- self._execute(
- "PUT",
- bucket_name,
- object_name=object_name,
- body=body,
- headers={"Content-MD5": cast(str, md5sum_hash(body))},
- query_params=cast(DictType, query_params),
- )
- def enable_object_legal_hold(
- self,
- bucket_name: str,
- object_name: str,
- version_id: Optional[str] = None,
- ):
- """
- Enable legal hold on an object.
- :param bucket_name: Name of the bucket.
- :param object_name: Object name in the bucket.
- :param version_id: Version ID of the object.
- Example::
- client.enable_object_legal_hold("my-bucket", "my-object")
- """
- check_bucket_name(bucket_name, s3_check=self._base_url.is_aws_host)
- check_object_name(object_name)
- body = marshal(LegalHold(True))
- query_params = {"versionId": version_id} if version_id else {}
- query_params["legal-hold"] = ""
- self._execute(
- "PUT",
- bucket_name,
- object_name=object_name,
- body=body,
- headers={"Content-MD5": cast(str, md5sum_hash(body))},
- query_params=cast(DictType, query_params),
- )
- def disable_object_legal_hold(
- self,
- bucket_name: str,
- object_name: str,
- version_id: Optional[str] = None,
- ):
- """
- Disable legal hold on an object.
- :param bucket_name: Name of the bucket.
- :param object_name: Object name in the bucket.
- :param version_id: Version ID of the object.
- Example::
- client.disable_object_legal_hold("my-bucket", "my-object")
- """
- check_bucket_name(bucket_name, s3_check=self._base_url.is_aws_host)
- check_object_name(object_name)
- body = marshal(LegalHold(False))
- query_params = {"versionId": version_id} if version_id else {}
- query_params["legal-hold"] = ""
- self._execute(
- "PUT",
- bucket_name,
- object_name=object_name,
- body=body,
- headers={"Content-MD5": cast(str, md5sum_hash(body))},
- query_params=cast(DictType, query_params),
- )
- def is_object_legal_hold_enabled(
- self,
- bucket_name: str,
- object_name: str,
- version_id: Optional[str] = None,
- ) -> bool:
- """
- Returns true if legal hold is enabled on an object.
- :param bucket_name: Name of the bucket.
- :param object_name: Object name in the bucket.
- :param version_id: Version ID of the object.
- Example::
- if client.is_object_legal_hold_enabled("my-bucket", "my-object"):
- print("legal hold is enabled on my-object")
- else:
- print("legal hold is not enabled on my-object")
- """
- check_bucket_name(bucket_name, s3_check=self._base_url.is_aws_host)
- check_object_name(object_name)
- query_params = {"versionId": version_id} if version_id else {}
- query_params["legal-hold"] = ""
- try:
- response = self._execute(
- "GET",
- bucket_name,
- object_name=object_name,
- query_params=cast(DictType, query_params),
- )
- legal_hold = unmarshal(LegalHold, response.data.decode())
- return legal_hold.status
- except S3Error as exc:
- if exc.code != "NoSuchObjectLockConfiguration":
- raise
- return False
- def delete_object_lock_config(self, bucket_name: str):
- """
- Delete object-lock configuration of a bucket.
- :param bucket_name: Name of the bucket.
- Example::
- client.delete_object_lock_config("my-bucket")
- """
- self.set_object_lock_config(
- bucket_name, ObjectLockConfig(None, None, None)
- )
- def get_object_lock_config(self, bucket_name: str) -> ObjectLockConfig:
- """
- Get object-lock configuration of a bucket.
- :param bucket_name: Name of the bucket.
- :return: :class:`ObjectLockConfig <ObjectLockConfig>` object.
- Example::
- config = client.get_object_lock_config("my-bucket")
- """
- check_bucket_name(bucket_name, s3_check=self._base_url.is_aws_host)
- response = self._execute(
- "GET", bucket_name, query_params={"object-lock": ""},
- )
- return unmarshal(ObjectLockConfig, response.data.decode())
- def set_object_lock_config(
- self,
- bucket_name: str,
- config: ObjectLockConfig,
- ):
- """
- Set object-lock configuration to a bucket.
- :param bucket_name: Name of the bucket.
- :param config: :class:`ObjectLockConfig <ObjectLockConfig>` object.
- Example::
- config = ObjectLockConfig(GOVERNANCE, 15, DAYS)
- client.set_object_lock_config("my-bucket", config)
- """
- check_bucket_name(bucket_name, s3_check=self._base_url.is_aws_host)
- if not isinstance(config, ObjectLockConfig):
- raise ValueError("config must be ObjectLockConfig type")
- body = marshal(config)
- self._execute(
- "PUT",
- bucket_name,
- body=body,
- headers={"Content-MD5": cast(str, md5sum_hash(body))},
- query_params={"object-lock": ""},
- )
- def get_object_retention(
- self,
- bucket_name: str,
- object_name: str,
- version_id: Optional[str] = None,
- ) -> Optional[Retention]:
- """
- Get retention configuration of an object.
- :param bucket_name: Name of the bucket.
- :param object_name: Object name in the bucket.
- :param version_id: Version ID of the object.
- :return: :class:`Retention <Retention>` object.
- Example::
- config = client.get_object_retention("my-bucket", "my-object")
- """
- check_bucket_name(bucket_name, s3_check=self._base_url.is_aws_host)
- check_object_name(object_name)
- query_params = {"versionId": version_id} if version_id else {}
- query_params["retention"] = ""
- try:
- response = self._execute(
- "GET",
- bucket_name,
- object_name=object_name,
- query_params=cast(DictType, query_params),
- )
- return unmarshal(Retention, response.data.decode())
- except S3Error as exc:
- if exc.code != "NoSuchObjectLockConfiguration":
- raise
- return None
- def set_object_retention(
- self,
- bucket_name: str,
- object_name: str,
- config: Retention,
- version_id: Optional[str] = None,
- ):
- """
- Set retention configuration on an object.
- :param bucket_name: Name of the bucket.
- :param object_name: Object name in the bucket.
- :param version_id: Version ID of the object.
- :param config: :class:`Retention <Retention>` object.
- Example::
- config = Retention(
- GOVERNANCE, datetime.utcnow() + timedelta(days=10),
- )
- client.set_object_retention("my-bucket", "my-object", config)
- """
- check_bucket_name(bucket_name, s3_check=self._base_url.is_aws_host)
- check_object_name(object_name)
- if not isinstance(config, Retention):
- raise ValueError("config must be Retention type")
- body = marshal(config)
- query_params = {"versionId": version_id} if version_id else {}
- query_params["retention"] = ""
- self._execute(
- "PUT",
- bucket_name,
- object_name=object_name,
- body=body,
- headers={"Content-MD5": cast(str, md5sum_hash(body))},
- query_params=cast(DictType, query_params),
- )
- def upload_snowball_objects(
- self,
- bucket_name: str,
- object_list: Iterable[SnowballObject],
- metadata: Optional[DictType] = None,
- sse: Optional[Sse] = None,
- tags: Optional[Tags] = None,
- retention: Optional[Retention] = None,
- legal_hold: bool = False,
- staging_filename: Optional[str] = None,
- compression: bool = False,
- ) -> ObjectWriteResult:
- """
- Uploads multiple objects in a single put call. It is done by creating
- intermediate TAR file optionally compressed which is uploaded to S3
- service.
- :param bucket_name: Name of the bucket.
- :param object_list: An iterable containing
- :class:`SnowballObject <SnowballObject>` object.
- :param metadata: Any additional metadata to be uploaded along
- with your PUT request.
- :param sse: Server-side encryption.
- :param tags: :class:`Tags` for the object.
- :param retention: :class:`Retention` configuration object.
- :param legal_hold: Flag to set legal hold for the object.
- :param staging_filename: A staging filename to create intermediate
- tarball.
- :param compression: Flag to compress TAR ball.
- :return: :class:`ObjectWriteResult` object.
- Example::
- # Upload snowball object.
- result = client.upload_snowball_objects(
- "my-bucket",
- [
- SnowballObject("my-object1", filename="/etc/hostname"),
- SnowballObject(
- "my-object2", data=io.BytesIO("hello"), length=5,
- ),
- SnowballObject(
- "my-object3", data=io.BytesIO("world"), length=5,
- mod_time=datetime.now(),
- ),
- ],
- )
- """
- check_bucket_name(bucket_name, s3_check=self._base_url.is_aws_host)
- object_name = f"snowball.{random()}.tar"
- # turn list like objects into an iterator.
- object_list = itertools.chain(object_list)
- metadata = metadata or {}
- metadata["X-Amz-Meta-Snowball-Auto-Extract"] = "true"
- name = staging_filename
- fileobj = None if name else BytesIO()
- with tarfile.open(
- name=name, mode="w:gz" if compression else "w", fileobj=fileobj,
- ) as tar:
- for obj in object_list:
- if obj.filename:
- tar.add(obj.filename, obj.object_name)
- else:
- info = tarfile.TarInfo(obj.object_name)
- info.size = cast(int, obj.length)
- info.mtime = int(
- time.to_float(obj.mod_time or time.utcnow()),
- )
- tar.addfile(info, obj.data)
- if not name:
- length = cast(BytesIO, fileobj).tell()
- cast(BytesIO, fileobj).seek(0)
- else:
- length = os.stat(name).st_size
- part_size = 0 if length < MIN_PART_SIZE else length
- if name:
- return self.fput_object(
- bucket_name,
- object_name,
- cast(str, staging_filename),
- metadata=metadata,
- sse=sse,
- tags=tags,
- retention=retention,
- legal_hold=legal_hold,
- part_size=part_size,
- )
- return self.put_object(
- bucket_name,
- object_name,
- cast(BinaryIO, fileobj),
- length,
- metadata=cast(Union[DictType, None], metadata),
- sse=sse,
- tags=tags,
- retention=retention,
- legal_hold=legal_hold,
- part_size=part_size,
- )
- def _list_objects(
- self,
- bucket_name: str,
- continuation_token: Optional[str] = None, # listV2 only
- delimiter: Optional[str] = None, # all
- encoding_type: Optional[str] = None, # all
- fetch_owner: Optional[bool] = None, # listV2 only
- include_user_meta: bool = False, # MinIO specific listV2.
- max_keys: Optional[int] = None, # all
- prefix: Optional[str] = None, # all
- start_after: Optional[str] = None,
- # all: v1:marker, versioned:key_marker
- version_id_marker: Optional[str] = None, # versioned
- use_api_v1: bool = False,
- include_version: bool = False,
- extra_headers: Optional[DictType] = None,
- extra_query_params: Optional[DictType] = None,
- ) -> Iterator[Object]:
- """
- List objects optionally including versions.
- Note: Its required to send empty values to delimiter/prefix and 1000 to
- max-keys when not provided for server-side bucket policy evaluation to
- succeed; otherwise AccessDenied error will be returned for such
- policies.
- """
- check_bucket_name(bucket_name, s3_check=self._base_url.is_aws_host)
- if version_id_marker:
- include_version = True
- is_truncated = True
- while is_truncated:
- query = extra_query_params or {}
- if include_version:
- query["versions"] = ""
- elif not use_api_v1:
- query["list-type"] = "2"
- if not include_version and not use_api_v1:
- if continuation_token:
- query["continuation-token"] = continuation_token
- if fetch_owner:
- query["fetch-owner"] = "true"
- if include_user_meta:
- query["metadata"] = "true"
- query["delimiter"] = delimiter or ""
- if encoding_type:
- query["encoding-type"] = encoding_type
- query["max-keys"] = str(max_keys or 1000)
- query["prefix"] = prefix or ""
- if start_after:
- if include_version:
- query["key-marker"] = start_after
- elif use_api_v1:
- query["marker"] = start_after
- else:
- query["start-after"] = start_after
- if version_id_marker:
- query["version-id-marker"] = version_id_marker
- response = self._execute(
- "GET",
- bucket_name,
- query_params=cast(DictType, query),
- headers=extra_headers,
- )
- objects, is_truncated, start_after, version_id_marker = (
- parse_list_objects(response)
- )
- if not include_version:
- version_id_marker = None
- if not use_api_v1:
- continuation_token = start_after
- yield from objects
- def _list_multipart_uploads(
- self,
- bucket_name: str,
- delimiter: Optional[str] = None,
- encoding_type: Optional[str] = None,
- key_marker: Optional[str] = None,
- max_uploads: Optional[int] = None,
- prefix: Optional[str] = None,
- upload_id_marker: Optional[str] = None,
- extra_headers: Optional[DictType] = None,
- extra_query_params: Optional[DictType] = None,
- ) -> ListMultipartUploadsResult:
- """
- Execute ListMultipartUploads S3 API.
- :param bucket_name: Name of the bucket.
- :param delimiter: (Optional) Delimiter on listing.
- :param encoding_type: (Optional) Encoding type.
- :param key_marker: (Optional) Key marker.
- :param max_uploads: (Optional) Maximum upload information to fetch.
- :param prefix: (Optional) Prefix on listing.
- :param upload_id_marker: (Optional) Upload ID marker.
- :param extra_headers: (Optional) Extra headers for advanced usage.
- :param extra_query_params: (Optional) Extra query parameters for
- advanced usage.
- :return:
- :class:`ListMultipartUploadsResult <ListMultipartUploadsResult>`
- object
- """
- query_params = extra_query_params or {}
- query_params.update(
- {
- "uploads": "",
- "delimiter": delimiter or "",
- "max-uploads": str(max_uploads or 1000),
- "prefix": prefix or "",
- "encoding-type": "url",
- },
- )
- if encoding_type:
- query_params["encoding-type"] = encoding_type
- if key_marker:
- query_params["key-marker"] = key_marker
- if upload_id_marker:
- query_params["upload-id-marker"] = upload_id_marker
- response = self._execute(
- "GET",
- bucket_name,
- query_params=cast(DictType, query_params),
- headers=cast(Union[DictType, None], extra_headers),
- )
- return ListMultipartUploadsResult(response)
- def _list_parts(
- self,
- bucket_name: str,
- object_name: str,
- upload_id: str,
- max_parts: Optional[int] = None,
- part_number_marker: Optional[str] = None,
- extra_headers: Optional[DictType] = None,
- extra_query_params: Optional[DictType] = None,
- ) -> ListPartsResult:
- """
- Execute ListParts S3 API.
- :param bucket_name: Name of the bucket.
- :param object_name: Object name in the bucket.
- :param upload_id: Upload ID.
- :param max_parts: (Optional) Maximum parts information to fetch.
- :param part_number_marker: (Optional) Part number marker.
- :param extra_headers: (Optional) Extra headers for advanced usage.
- :param extra_query_params: (Optional) Extra query parameters for
- advanced usage.
- :return: :class:`ListPartsResult <ListPartsResult>` object
- """
- query_params = extra_query_params or {}
- query_params.update(
- {
- "uploadId": upload_id,
- "max-parts": str(max_parts or 1000),
- },
- )
- if part_number_marker:
- query_params["part-number-marker"] = part_number_marker
- response = self._execute(
- "GET",
- bucket_name,
- object_name=object_name,
- query_params=cast(DictType, query_params),
- headers=cast(Union[DictType, None], extra_headers),
- )
- return ListPartsResult(response)
|