cloudpathlib.S3Path¶
Class for representing and operating on AWS S3 URIs, in the style of the Python standard
library's pathlib
module. Instances
represent a path in S3 with filesystem path semantics, and convenient methods allow for basic
operations like joining, reading, writing, iterating over contents, etc. This class almost
entirely mimics the pathlib.Path
interface, so most familiar properties and methods should be available and behave in the
expected way.
The S3Client
class handles authentication with AWS. If a client instance is
not explicitly specified on S3Path
instantiation, a default client is used. See S3Client
's
documentation for more details.
Source code in cloudpathlib/s3/s3path.py
@register_path_class("s3")
class S3Path(CloudPath):
"""Class for representing and operating on AWS S3 URIs, in the style of the Python standard
library's [`pathlib` module](https://docs.python.org/3/library/pathlib.html). Instances
represent a path in S3 with filesystem path semantics, and convenient methods allow for basic
operations like joining, reading, writing, iterating over contents, etc. This class almost
entirely mimics the [`pathlib.Path`](https://docs.python.org/3/library/pathlib.html#pathlib.Path)
interface, so most familiar properties and methods should be available and behave in the
expected way.
The [`S3Client`](../s3client/) class handles authentication with AWS. If a client instance is
not explicitly specified on `S3Path` instantiation, a default client is used. See `S3Client`'s
documentation for more details.
"""
cloud_prefix: str = "s3://"
client: "S3Client"
@property
def drive(self) -> str:
return self.bucket
def mkdir(self, parents=False, exist_ok=False):
# not possible to make empty directory on s3
pass
def touch(self, exist_ok: bool = True):
if self.exists():
if not exist_ok:
raise FileExistsError(f"File exists: {self}")
self.client._move_file(self, self)
else:
tf = TemporaryDirectory()
p = Path(tf.name) / "empty"
p.touch()
self.client._upload_file(p, self)
tf.cleanup()
def stat(self):
try:
meta = self.client._get_metadata(self)
except self.client.client.exceptions.NoSuchKey:
raise NoStatError(
f"No stats available for {self}; it may be a directory or not exist."
)
return os.stat_result(
(
None, # mode
None, # ino
self.cloud_prefix, # dev,
None, # nlink,
None, # uid,
None, # gid,
meta.get("size", 0), # size,
None, # atime,
meta.get("last_modified", 0).timestamp(), # mtime,
None, # ctime,
)
)
@property
def bucket(self) -> str:
return self._no_prefix.split("/", 1)[0]
@property
def key(self) -> str:
key = self._no_prefix_no_drive
# key should never have starting slash for
# use with boto, etc.
if key.startswith("/"):
key = key[1:]
return key
@property
def etag(self):
return self.client._get_metadata(self).get("etag")
Attributes¶
bucket: str
property
readonly
¶
cloud_prefix: str
¶
drive: str
property
readonly
¶
The drive prefix (letter or UNC path), if any. (Docstring copied from pathlib.Path)
etag
property
readonly
¶
key: str
property
readonly
¶
Methods¶
mkdir(self, parents = False, exist_ok = False)
¶
Create a new directory at this given path. (Docstring copied from pathlib.Path)
Source code in cloudpathlib/s3/s3path.py
def mkdir(self, parents=False, exist_ok=False):
# not possible to make empty directory on s3
pass
stat(self)
¶
Return the result of the stat() system call on this path, like os.stat() does. (Docstring copied from pathlib.Path)
Source code in cloudpathlib/s3/s3path.py
def stat(self):
try:
meta = self.client._get_metadata(self)
except self.client.client.exceptions.NoSuchKey:
raise NoStatError(
f"No stats available for {self}; it may be a directory or not exist."
)
return os.stat_result(
(
None, # mode
None, # ino
self.cloud_prefix, # dev,
None, # nlink,
None, # uid,
None, # gid,
meta.get("size", 0), # size,
None, # atime,
meta.get("last_modified", 0).timestamp(), # mtime,
None, # ctime,
)
)
touch(self, exist_ok: bool = True)
¶
Create this file with the given access mode, if it doesn't exist. (Docstring copied from pathlib.Path)
Source code in cloudpathlib/s3/s3path.py
def touch(self, exist_ok: bool = True):
if self.exists():
if not exist_ok:
raise FileExistsError(f"File exists: {self}")
self.client._move_file(self, self)
else:
tf = TemporaryDirectory()
p = Path(tf.name) / "empty"
p.touch()
self.client._upload_file(p, self)
tf.cleanup()