rw

Read and write related API.

The methods read or write one S3 object: a file, or a folder marker object (e.g. folder/, which mkdir writes). A bucket, a relative path or the void path raise S3PathTypeError before calling AWS.

Writing returns a new S3Path with the metadata snapshot of the written object, and leaves this one as it was (see s3pathlib.core.metadata). Without version_id, reading reads the latest version.

Errors of the AWS calls are raised as is (botocore.exceptions.ClientError, check e.response["Error"]["Code"]), e.g. NoSuchKey, NoSuchBucket, PreconditionFailed (if_match / if_none_match not met when writing, if_match not met when reading), 304 (if_none_match met when reading).

Design notes

``touch`` and ``mkdir`` are conditional writes. They write with IfNoneMatch="*": one atomic call that never replaces an existing object. A head_object followed by a put_object could replace an object that someone else wrote in between.

Writing returns a new path. The snapshot of the written object is a partial one built from the put_object response, so size and metadata need no extra call, and this S3Path keeps describing what it described.

class s3pathlib.core.rw.ReadAndWriteAPIMixin[source]

A mixin class that implements the Text / Bytes, Read / Write methods.

read_bytes(version_id: str = OPT, if_match: str = OPT, if_modified_since: datetime = OPT, if_none_match: str = OPT, if_unmodified_since: datetime = OPT, range: str = OPT, response_cache_control: str = OPT, response_content_disposition: str = OPT, response_content_encoding: str = OPT, response_content_language: str = OPT, response_content_type: str = OPT, response_expires: str = OPT, sse_customer_algorithm: str = OPT, sse_customer_key: str = OPT, request_payer: str = OPT, part_number: int = OPT, expected_bucket_owner: str = OPT, checksum_mode: str = OPT, s3_client: S3Client | None = None) → bytes[source]

Read the content of the object, a wrapper of get_object.

If this S3Path has no metadata snapshot yet, the get_object response becomes its snapshot (etag, size, version_id, …), unless range, part_number or a response_* argument is given (the response then describes a part, or has overridden headers). An existing snapshot is never replaced, see s3pathlib.core.metadata.

Example:

>>> s3path = S3Path("s3://my-bucket/read-bytes.txt")
>>> _ = s3path.write_bytes(b"hello", metadata={"creator": "me"})
>>> s3path.read_bytes()
b'hello'
>>> s3path.size
5
>>> s3path.metadata
{'creator': 'me'}
Returns:

the binary data.

Raises:

S3PathTypeError – for a bucket, a relative path or the void path.

read_text(encoding='utf-8', errors='strict', version_id: str = OPT, if_match: str = OPT, if_modified_since: datetime = OPT, if_none_match: str = OPT, if_unmodified_since: datetime = OPT, range: str = OPT, response_cache_control: str = OPT, response_content_disposition: str = OPT, response_content_encoding: str = OPT, response_content_language: str = OPT, response_content_type: str = OPT, response_expires: str = OPT, sse_customer_algorithm: str = OPT, sse_customer_key: str = OPT, request_payer: str = OPT, part_number: int = OPT, expected_bucket_owner: str = OPT, checksum_mode: str = OPT, s3_client: S3Client | None = None) → str[source]

Read the content of the object as text, see read_bytes(), which also tells when the metadata snapshot is filled.

Example:

>>> s3path = S3Path("s3://my-bucket/read-text.txt")
>>> _ = s3path.write_text("hello")
>>> s3path.read_text()
'hello'
Parameters:
Returns:

the string data.

Raises:

S3PathTypeError – for a bucket, a relative path or the void path.

write_bytes(data: bytes, metadata: MetadataType = OPT, tags: TagType = OPT, acl: str = OPT, cache_control: str = OPT, content_disposition: str = OPT, content_encoding: str = OPT, content_language: str = OPT, content_length: int = OPT, content_md5: str = OPT, content_type: str = OPT, checksum_algorithm: str = OPT, checksum_crc32: str = OPT, checksum_crc32c: str = OPT, checksum_sha1: str = OPT, checksum_sha256: str = OPT, expires_datetime: datetime = OPT, grant_full_control: str = OPT, grant_read: str = OPT, grant_read_acp: str = OPT, grant_write_acp: str = OPT, server_side_encryption: str = OPT, storage_class: str = OPT, website_redirect_location: str = OPT, sse_customer_algorithm: str = OPT, sse_customer_key: str = OPT, sse_kms_key_id: str = OPT, sse_kms_encryption_context: str = OPT, bucket_key_enabled: bool = OPT, request_payer: str = OPT, object_lock_mode: str = OPT, object_lock_retain_until_datetime: datetime = OPT, object_lock_legal_hold_status: str = OPT, expected_bucket_owner: str = OPT, if_match: str = OPT, if_none_match: str = OPT, s3_client: S3Client | None = None) → S3Path[source]

Write data to the object (replacing it if it exists), a wrapper of put_object.

Example:

>>> s3path = S3Path("s3://my-bucket/write-bytes.txt")
>>> s3path_new = s3path.write_bytes(b"hello", metadata={"creator": "me"})
>>> s3path_new.size
5
>>> s3path_new.metadata
{'creator': 'me'}
Parameters:

  • data – the content to write.

  • metadata – the user metadata, a dict of str. S3 stores the keys in lower case.

  • tags – the object tags, a dict of str.

  • if_match – only write if the ETag of the object is this one (nobody changed it since you read it).

  • if_none_match – "*" only writes if the object does not exist (a delete marker as the latest version counts as not existing).

Returns:

a new S3Path with the metadata snapshot of the written object (size, etag, metadata, version_id, …); this S3Path is not changed.

Raises:

S3PathTypeError – for a bucket, a relative path or the void path.

write_text(data: str, encoding: str = 'utf-8', errors: str = 'strict', metadata: MetadataType = OPT, tags: TagType = OPT, acl: str = OPT, cache_control: str = OPT, content_disposition: str = OPT, content_encoding: str = OPT, content_language: str = OPT, content_length: int = OPT, content_md5: str = OPT, content_type: str = OPT, checksum_algorithm: str = OPT, checksum_crc32: str = OPT, checksum_crc32c: str = OPT, checksum_sha1: str = OPT, checksum_sha256: str = OPT, expires_datetime: datetime = OPT, grant_full_control: str = OPT, grant_read: str = OPT, grant_read_acp: str = OPT, grant_write_acp: str = OPT, server_side_encryption: str = OPT, storage_class: str = OPT, website_redirect_location: str = OPT, sse_customer_algorithm: str = OPT, sse_customer_key: str = OPT, sse_kms_key_id: str = OPT, sse_kms_encryption_context: str = OPT, bucket_key_enabled: bool = OPT, request_payer: str = OPT, object_lock_mode: str = OPT, object_lock_retain_until_datetime: datetime = OPT, object_lock_legal_hold_status: str = OPT, expected_bucket_owner: str = OPT, if_match: str = OPT, if_none_match: str = OPT, s3_client: S3Client | None = None) → S3Path[source]

Write the text data to the object, see write_bytes().

Example:

>>> s3path_new = S3Path("s3://my-bucket/write-text.txt").write_text("hello")
>>> s3path_new.size
5
Parameters:

  • data – the text to write.

  • encoding – See encode.

  • errors – See encode.

  • metadata – the user metadata, a dict of str. S3 stores the keys in lower case.

  • tags – the object tags, a dict of str.

  • if_match – only write if the ETag of the object is this one (nobody changed it since you read it).

  • if_none_match – "*" only writes if the object does not exist (a delete marker as the latest version counts as not existing).

Returns:

a new S3Path with the metadata snapshot of the written object (size, etag, metadata, version_id, …); this S3Path is not changed.

Raises:

S3PathTypeError – for a bucket, a relative path or the void path.

touch(exist_ok: bool = False, metadata: MetadataType = OPT, tags: TagType = OPT, acl: str = OPT, cache_control: str = OPT, content_disposition: str = OPT, content_encoding: str = OPT, content_language: str = OPT, content_length: int = OPT, content_md5: str = OPT, content_type: str = OPT, checksum_algorithm: str = OPT, checksum_crc32: str = OPT, checksum_crc32c: str = OPT, checksum_sha1: str = OPT, checksum_sha256: str = OPT, expires_datetime: datetime = OPT, grant_full_control: str = OPT, grant_read: str = OPT, grant_read_acp: str = OPT, grant_write_acp: str = OPT, server_side_encryption: str = OPT, storage_class: str = OPT, website_redirect_location: str = OPT, sse_customer_algorithm: str = OPT, sse_customer_key: str = OPT, sse_kms_key_id: str = OPT, sse_kms_encryption_context: str = OPT, bucket_key_enabled: bool = OPT, request_payer: str = OPT, object_lock_mode: str = OPT, object_lock_retain_until_datetime: datetime = OPT, object_lock_legal_hold_status: str = OPT, expected_bucket_owner: str = OPT, s3_client: S3Client | None = None) → None[source]

Create an empty object if it does not exist, with one conditional put_object (IfNoneMatch="*"), so it is atomic: an object written by someone else in between is never replaced. An existing object is never changed.

Example:

>>> s3path = S3Path("s3://my-bucket/touch.txt")
>>> s3path.touch()
>>> s3path.read_text()
''
>>> s3path.touch(exist_ok=True)
Parameters:

  • exist_ok – if False (default), raise S3FileAlreadyExist when the object exists; if True, do nothing then.

  • metadata – the user metadata of the new object.

  • tags – the tags of the new object.

The other arguments are the ones of write_bytes().

Raises:
mkdir(exist_ok: bool = False, parents: bool = False, s3_client: S3Client | None = None) → None[source]

Create the folder marker object of this folder (an empty object whose key ends with /, what “Create folder” does in the S3 console), and of each parent folder with parents=True. A marker that exists is never replaced (conditional put_object with IfNoneMatch="*").

A folder exists in S3 as soon as an object key starts with it, even without marker (“soft folder”), see exists().

Example:

>>> s3dir = S3Path("s3://my-bucket/mkdir/sub/")
>>> s3dir.mkdir(parents=True)
>>> S3Path("s3://my-bucket/mkdir/").read_text()
''
>>> s3dir.mkdir(exist_ok=True)
Parameters:

  • exist_ok – if False (default), raise S3FolderAlreadyExist when the folder exists (with or without marker); if True, only write the markers that do not exist.

  • parents – if True, also write the markers of the parent folders (up to the bucket).

Raises: