S3Store
Objects in a bucket of an S3-compatible service, addressed path-style.
Requests are signed with AWS Signature Version 4. Conditional writes use
If-Match and If-None-Match, which the service must honor.
Attributes
attributeendpoint= endpoint.rstrip('/')attributebucket= bucketattributeprefix= prefix.strip('/') + '/' if prefix.strip('/') else ''attributeaccess_key= access_keyattributesecret_key= secret_keyattributeregion= regionFunctions
constructor__init__(endpoint, bucket, prefix, access_key, secret_key, region='us-east-1', transport=None) -> NoneBind the store to a bucket.
paramendpointstrBase URL of the service, https://s3.example.com.
parambucketstrBucket name.
paramprefixstrKey prefix every object is kept under; may be empty.
paramaccess_keystrAccess key id.
paramsecret_keystrSecret access key.
paramregionstr= 'us-east-1'Region the signature names.
paramtransporthttpx.BaseTransport | None= NoneTransport override for tests.
Returns
Nonefunc_sign(method, path, query, headers, body) -> NoneAdd the Signature Version 4 headers for a request to headers.
parammethodstrparampathstrparamquerydict[str, str]paramheadersdict[str, str]parambodybytesReturns
Nonefunc_request(method, key, query=None, headers=None, body=b'') -> httpx.ResponseSend a signed request for an object, or for the bucket when key is None.
parammethodstrparamkeystr | Noneparamquerydict[str, str] | None= Noneparamheadersdict[str, str] | None= Noneparambodybytes= b''Returns
httpx.httpx.Responsefunc_check(response, key) -> NoneRaise for a refused condition or any other failure.
paramresponsehttpx.ResponseparamkeystrReturns
Nonefuncget(key) -> tuple[bytes, str] | NoneFetch an object.
paramkeystrReturns
tuple[bytes, str] | Nonefuncput(key, data, if_match=None, if_absent=False) -> strUpload an object, conditionally when asked.
paramkeystrparamdatabytesparamif_matchstr | None= Noneparamif_absentbool= FalseReturns
strfuncdelete(key, if_match=None) -> NoneDelete an object, after comparing its entity tag when asked.
paramkeystrparamif_matchstr | None= NoneReturns
Nonefunclist(prefix='') -> dict[str, str]List the bucket under the store's prefix, following continuation tokens.
paramprefixstr= ''Returns
dict[str, str]