From e12d5d584c0f5b6f6ff7944585ca66941be18e1a Mon Sep 17 00:00:00 2001 From: laughingman7743 Date: Mon, 5 Oct 2026 01:16:31 +0900 Subject: [PATCH] Move the bucket versioning request into S3Core #1084 added a GetBucketVersioning request built in S3FileSystem and a call to the private S3Core._is_directory_bucket(). As agreed on #1086, add S3Core.get_bucket_versioning(), which returns the versioning status or None, and the public S3Path.is_directory_bucket property, which also replaces the private helper in plan_multipart_copy(). The aio mv tests that mocked only the sync filesystem's _call now share the mock with its core, as the other tests do, so that they keep observing the request. Co-Authored-By: Claude Opus 5.5 --- docs/filesystem.md | 3 ++ pyathena/filesystem/s3.py | 9 ++---- pyathena/filesystem/s3_core.py | 35 ++++++++++++---------- pyathena/filesystem/s3_path.py | 8 +++++ tests/pyathena/filesystem/test_s3_async.py | 10 ++++--- tests/pyathena/filesystem/test_s3_core.py | 30 +++++++++++++++++++ tests/pyathena/filesystem/test_s3_path.py | 12 ++++++++ 7 files changed, 80 insertions(+), 27 deletions(-) diff --git a/docs/filesystem.md b/docs/filesystem.md index 585cbc59..dd3d54ad 100644 --- a/docs/filesystem.md +++ b/docs/filesystem.md @@ -401,6 +401,9 @@ upload_url = core.generate_presigned_url(path, "put_object", ContentType="text/c `create_bucket()` and `delete_bucket()` create and delete a bucket when called. The `allow_bucket_creation` and `allow_bucket_deletion` options apply only to the filesystem's `mkdir`/`makedirs` and `rmdir`, not to calls through the core. +`get_bucket_versioning()` returns the versioning state of a bucket, `Enabled` or +`Suspended`, or `None` if versioning has never been enabled on it. +`S3Path.is_directory_bucket` tells whether the bucket of a path is a directory bucket. `create_bucket()` sends a `LocationConstraint` for the `region_name` argument, or for the client's region by default, except in `us-east-1`. The fields of a diff --git a/pyathena/filesystem/s3.py b/pyathena/filesystem/s3.py index e6d04f47..803f3f21 100644 --- a/pyathena/filesystem/s3.py +++ b/pyathena/filesystem/s3.py @@ -1438,15 +1438,10 @@ def _move_pairs( buckets = { p.bucket for p in paths.values() - if p.version_id == "null" - and p.name in unversioned - and not self.core._is_directory_bucket(p.bucket) + if p.version_id == "null" and p.name in unversioned and not p.is_directory_bucket } versioned_buckets = { - bucket - for bucket in buckets - if self._call(self._client.get_bucket_versioning, Bucket=bucket).get("Status") - == "Enabled" + bucket for bucket in buckets if self.core.get_bucket_versioning(bucket) == "Enabled" } missing = { source diff --git a/pyathena/filesystem/s3_core.py b/pyathena/filesystem/s3_core.py index 41f4e768..adb940eb 100644 --- a/pyathena/filesystem/s3_core.py +++ b/pyathena/filesystem/s3_core.py @@ -835,6 +835,23 @@ def delete_bucket(self, bucket: str, **params) -> None: _logger.debug(f"Delete bucket: s3://{bucket}") self.call(self._client.delete_bucket, Bucket=bucket, **params) + def get_bucket_versioning(self, bucket: str, **params) -> str | None: + """Get the versioning state of a bucket with GetBucketVersioning. + + Args: + bucket: The name of the bucket. + **params: Additional request parameters, sent as given. + + Returns: + ``Enabled`` or ``Suspended``, or None if versioning has never + been enabled on the bucket. + + Raises: + FileNotFoundError: If the bucket does not exist. + """ + response = self.call(self._client.get_bucket_versioning, Bucket=bucket, **params) + return cast(str | None, response.get("Status")) + def delete_object(self, path: S3Path, **params) -> None: """Delete an object, or a version of it, with DeleteObject. @@ -1266,7 +1283,7 @@ def plan_multipart_copy( request.pop("Tagging", None) # Directory buckets do not support GetObjectTagging, and their # objects have no tags. - if not self._is_directory_bucket(source.bucket): + if not source.is_directory_bucket: tags = self.get_object_tagging( source, **self.operation_params("get_object_tagging", source_params) ) @@ -1290,7 +1307,7 @@ def plan_multipart_copy( ) if annotation_directive == "COPY" and "CopySourceSSECustomerAlgorithm" not in params - and not self._is_directory_bucket(source.bucket) + and not source.is_directory_bucket else () ) return S3MultipartCopyPlan( @@ -1604,20 +1621,6 @@ def generate_presigned_url( ), ) - @staticmethod - def _is_directory_bucket(bucket: str) -> bool: - """Return whether the bucket is a directory bucket (S3 Express One Zone). - - Directory bucket names end with ``--x-s3``. - - Args: - bucket: S3 bucket name. - - Returns: - True if the bucket is a directory bucket. - """ - return bucket.endswith("--x-s3") - @staticmethod def _copy_source_params(params: Mapping[str, Any]) -> dict[str, Any]: """Map the parameters of a copy to those of the requests that read its source. diff --git a/pyathena/filesystem/s3_path.py b/pyathena/filesystem/s3_path.py index 19dbcf84..54c8f2d0 100644 --- a/pyathena/filesystem/s3_path.py +++ b/pyathena/filesystem/s3_path.py @@ -123,6 +123,14 @@ def is_bucket(self) -> bool: """Whether the path names the bucket: it has no key, or a key of only slashes.""" return not self.key or not self.key.strip("/") + @property + def is_directory_bucket(self) -> bool: + """Whether the bucket is a directory bucket (S3 Express One Zone). + + The names of directory buckets end with ``--x-s3``. + """ + return self.bucket.endswith("--x-s3") + @property def name(self) -> str: """The path without a scheme or version, in ``bucket/key`` form, or the bucket.""" diff --git a/tests/pyathena/filesystem/test_s3_async.py b/tests/pyathena/filesystem/test_s3_async.py index 96064b77..c504a5e1 100644 --- a/tests/pyathena/filesystem/test_s3_async.py +++ b/tests/pyathena/filesystem/test_s3_async.py @@ -1036,7 +1036,7 @@ async def copy_file(path1, path2, **kwargs): @pytest.mark.asyncio async def test_mv_null_version_onto_key(self, status): fs = AioS3FileSystem(connection=mock.MagicMock(), skip_instance_cache=True) - fs._sync_fs._call = mock.MagicMock( + fs._sync_fs._call = fs._sync_fs._core.call = mock.MagicMock( return_value={} if status is None else {"Status": status} ) fs._copy_file = mock.AsyncMock(return_value=True) @@ -1063,7 +1063,7 @@ async def test_mv_null_version_onto_key(self, status): @pytest.mark.asyncio async def test_mv_null_version_conflicts_depend_on_bucket_state(self, status): fs = AioS3FileSystem(connection=mock.MagicMock(), skip_instance_cache=True) - fs._sync_fs._call = mock.MagicMock( + fs._sync_fs._call = fs._sync_fs._core.call = mock.MagicMock( return_value={} if status is None else {"Status": status} ) fs._copy_file = mock.AsyncMock(return_value=True) @@ -1090,7 +1090,7 @@ async def test_mv_null_version_conflicts_depend_on_bucket_state(self, status): @pytest.mark.asyncio async def test_mv_null_version_directory_bucket_does_not_read_bucket_state(self): fs = AioS3FileSystem(connection=mock.MagicMock(), skip_instance_cache=True) - fs._sync_fs._call = mock.MagicMock() + fs._sync_fs._call = fs._sync_fs._core.call = mock.MagicMock() fs._copy_file = mock.AsyncMock(return_value=True) fs._delete_objects = mock.AsyncMock() key = "s3://example--usw2-az1--x-s3/key" @@ -1105,7 +1105,9 @@ async def test_mv_null_version_directory_bucket_does_not_read_bucket_state(self) @pytest.mark.asyncio async def test_mv_null_version_failure_does_not_delete(self, stage): fs = AioS3FileSystem(connection=mock.MagicMock(), skip_instance_cache=True) - fs._sync_fs._call = mock.MagicMock(return_value={"Status": "Enabled"}) + fs._sync_fs._call = fs._sync_fs._core.call = mock.MagicMock( + return_value={"Status": "Enabled"} + ) fs._copy_file = mock.AsyncMock(return_value=True) fs._delete_objects = mock.AsyncMock() if stage == "lookup": diff --git a/tests/pyathena/filesystem/test_s3_core.py b/tests/pyathena/filesystem/test_s3_core.py index b33b58cb..ef992573 100644 --- a/tests/pyathena/filesystem/test_s3_core.py +++ b/tests/pyathena/filesystem/test_s3_core.py @@ -425,6 +425,36 @@ def test_delete_bucket(self): core.delete_bucket("bucket") stubber.assert_no_pending_responses() + @pytest.mark.parametrize( + ("response", "expected"), + [ + ({"Status": "Enabled"}, "Enabled"), + ({"Status": "Suspended", "MFADelete": "Disabled"}, "Suspended"), + # A bucket whose versioning has never been enabled. + ({}, None), + ], + ) + def test_get_bucket_versioning(self, response, expected): + core, stubber = _make_core(request_kwargs={"RequestPayer": "requester"}) + # RequestPayer, which GetBucketVersioning does not accept, is not sent. + stubber.add_response( + "get_bucket_versioning", + response, + {"Bucket": "bucket", "ExpectedBucketOwner": "123456789012"}, + ) + with stubber: + status = core.get_bucket_versioning("bucket", ExpectedBucketOwner="123456789012") + stubber.assert_no_pending_responses() + assert status == expected + + def test_get_bucket_versioning_translates_errors(self): + core, stubber = _make_core() + stubber.add_client_error( + "get_bucket_versioning", service_error_code="NoSuchBucket", http_status_code=404 + ) + with stubber, pytest.raises(FileNotFoundError): + core.get_bucket_versioning("bucket") + def test_list_objects(self): core, stubber = _make_core() stubber.add_response( diff --git a/tests/pyathena/filesystem/test_s3_path.py b/tests/pyathena/filesystem/test_s3_path.py index 0dda470d..aac76dcc 100644 --- a/tests/pyathena/filesystem/test_s3_path.py +++ b/tests/pyathena/filesystem/test_s3_path.py @@ -65,6 +65,18 @@ def test_parse_invalid(self, path): def test_is_bucket(self, path, expected): assert path.is_bucket is expected + @pytest.mark.parametrize( + ("path", "expected"), + [ + (S3Path("bucket--usw2-az1--x-s3", "key"), True), + (S3Path("bucket--usw2-az1--x-s3"), True), + (S3Path("bucket", "key--x-s3"), False), + (S3Path("bucket--x-s3-other"), False), + ], + ) + def test_is_directory_bucket(self, path, expected): + assert path.is_directory_bucket is expected + @pytest.mark.parametrize( ("path", "name", "string", "uri"), [