add support for S3 object version IDs

s3path.py

@@ -189,6 +189,15 @@ def stat(self, path, *, follow_symlinks=True):
             raise NotImplementedError(
                 f'Setting follow_symlinks to {follow_symlinks} is unsupported on S3 service.')
         resource, _ = self.configuration_map.get_configuration(path)
+
+        if hasattr(path, 'version_id') and path.version_id is not None:
+            object_summary = resource.ObjectVersion(path.bucket, path.key, path.version_id).get()
+            return StatResult(
+                size=object_summary['ContentLength'],
+                last_modified=object_summary['LastModified'],
+                version_id=object_summary['VersionId'],
+            )
+
         object_summary = resource.ObjectSummary(path.bucket, path.key)
         return StatResult(
             size=object_summary.size,
@@ -245,6 +254,8 @@ def open(self, path, *, mode='r', buffering=-1, encoding=None, errors=None, newl
             'newline': newline,
         }
         transport_params = {'defer_seek': True}
+        if hasattr(path, 'version_id') and path.version_id is not None:
+            transport_params['version_id'] = path.version_id
         dummy_object = resource.Object('bucket', 'key')
         if smart_open.__version__ >= '5.1.0':
             self._smart_open_new_version_kwargs(
@@ -764,7 +775,20 @@ def from_uri(cls, uri):
         """
         if not uri.startswith('s3://'):
             raise ValueError('Provided uri seems to be no S3 URI!')
-        return cls(uri[4:])
+
+        uri, *version_id = uri.split('?VersionID=')
+
+        if len(version_id) > 1:
+            raise ValueError('Do you use "?VersionID=" in your bucket or key?')
+        elif len(version_id) == 0:
+            version_id = None
+        else:
+            version_id = version_id[0]
+
+        self = cls(uri[4:])
+        self.version_id = version_id
+
+        return self
 
     @property
     def bucket(self):
@@ -808,7 +832,22 @@ def from_bucket_key(cls, bucket, key):
         key = cls(key)
         if key.is_absolute():
             key = key.relative_to('/')
-        return bucket / key
+        self = bucket / key
+        self.version_id = None
+        return self
+
+    @classmethod
+    def from_bucket_key_versionid(cls, bucket, key, version_id):
+        """
+        from_bucket_key_versionid class method create a class instance from bucket, key, and version_id 
+
+        >> from s3path import PureS3Path
+        >> PureS3Path.from_bucket_key_versionid(bucket='<bucket>', key='<key>', version_id='<version_id>')
+        << PureS3Path('/<bucket>/<key>')
+        """
+        self = cls.from_bucket_key(bucket, key)
+        self.version_id = version_id
+        return self
 
     def as_uri(self):
         """
@@ -833,6 +872,21 @@ class S3Path(_PathNotSupportedMixin, Path, PureS3Path):
     _accessor = _s3_accessor
     __slots__ = ()
 
+    def __new__(cls, *args, **kwargs):
+
+        args, *version_id = args.split('?VersionID=')
+
+        if len(version_id) > 1:
+            raise ValueError('Do you use "?VersionID=" in your bucket or key?')
+        elif len(version_id) == 0:
+            version_id = None
+        else:
+            version_id = version_id[0]
+
+        self = super().__new__(cls, *args, **kwargs)
+        self.version_id = version_id
+        return self
+
     def _init(self, template=None):
         super()._init(template)
         if template is None:
@@ -1124,7 +1178,7 @@ def absolute(self):
         raise ValueError("Absolute path can't be determined for relative S3Path objects")
 
 
-class StatResult(namedtuple('BaseStatResult', 'size, last_modified')):
+class StatResult(namedtuple('BaseStatResult', 'size, last_modified, version_id', defaults=(None,))):
     """
     Base of os.stat_result but with boto3 s3 features
     """
@@ -1142,12 +1196,16 @@ def st_size(self):
     def st_mtime(self):
         return self.last_modified.timestamp()
 
+    @property
+    def st_version_id(self):
+        return self.version_id
+
 
 class S3DirEntry:
-    def __init__(self, name, is_dir, size=None, last_modified=None):
+    def __init__(self, name, is_dir, size=None, last_modified=None, version_id=None):
         self.name = name
         self._is_dir = is_dir
-        self._stat = StatResult(size=size, last_modified=last_modified)
+        self._stat = StatResult(size=size, last_modified=last_modified, version_id=version_id)
 
     def __repr__(self):
         return f'{type(self).__name__}(name={self.name}, is_dir={self._is_dir}, stat={self._stat})'

tests/test_path_operations.py

@@ -704,4 +704,42 @@ def test_absolute(s3_mock):
 
     relative_path = S3Path('./Test.test')
     with pytest.raises(ValueError):
-        relative_path.absolute()
+        relative_path.absolute()
+
+
+def test_versioned_bucket(s3_mock):
+    bucket, key = 'test-versioned-bucket', 'versioned_file.txt'
+
+    s3 = boto3.resource('s3')
+    s3.create_bucket(Bucket=bucket)
+    s3.BucketVersioning(bucket).enable()
+
+    object_summary = s3.ObjectSummary(bucket, key)
+    file_contents_by_version = (b'Test', b'Test updated', b'Test', b'Test final')
+
+    version_id_to_file_content = {}
+    for file_content in file_contents_by_version:
+        version_id = object_summary.put(Body=file_content).get("VersionId")
+        version_id_to_file_content[version_id] = file_content
+
+    assert len(version_id_to_file_content) == 4
+
+    def assert_expected_file_content(s3_paths, expected_file_content):
+        for s3_path in s3_paths:
+            with s3_path.open(mode='rb') as file_pointer:
+                assert file_pointer.read() == expected_file_content
+
+    # Test that we can read specific versions of the file
+    for version_id, file_content in version_id_to_file_content.items():
+        paths = (
+            S3Path.from_uri(f's3://{bucket}/{key}?VersionID={version_id}'),
+            S3Path.from_bucket_key_versionid(bucket=bucket, key=key, version_id=version_id),
+        )
+        assert_expected_file_content(s3_paths=paths, expected_file_content=file_content)
+
+    # Test that we receive the latest version of the file when no version_id is specified
+    paths = (
+        S3Path.from_uri(f's3://{bucket}/{key}'),
+        S3Path.from_bucket_key(bucket=bucket, key=key),
+    )
+    assert_expected_file_content(s3_paths=paths, expected_file_content=file_contents_by_version[-1])