Merge pull request #1090 from pyathena-dev/refactor/1086-tags-acl-metadata

Move tagging, ACL, metadata and presign requests into S3Core

Author: laughingman7743

docs/filesystem.md

@@ -300,13 +300,15 @@ directories below the bucket level) and is always a no-op.
 ## Typed S3 operations
 
 `S3FileSystem.core` is an `S3Core`, the typed operations that the filesystem sends
-its listing, lookup, read, write, delete, multipart upload and copy requests with. It can also be
-built on a boto3 S3 client. Each operation sends one request (one per page for the
-iterators and `list_object_annotations()`); `plan_multipart_copy()` and
-`copy_object_annotation()`, described below, send several. The requests are sent with
-the retry policy. An operation raises `FileNotFoundError` for a missing bucket or
-multipart upload, or for a missing object or version that it reads, and caches
-nothing. Requests sent through `fs.core` do not invalidate the filesystem's cache: call
+its listing, lookup, read, write, delete, multipart upload, copy, tagging, ACL and
+metadata replacement requests with. It can also be built on a boto3 S3 client. Each
+operation sends one request (one per page for the iterators and
+`list_object_annotations()`); `plan_multipart_copy()` and `copy_object_annotation()`,
+described below, send several, and `generate_presigned_url()` signs a URL locally
+without a request. The requests are sent with the retry policy. An operation raises
+`FileNotFoundError` for a missing bucket or multipart upload, or for a missing object or
+version that it reads, and caches nothing. Requests sent through `fs.core` do not
+invalidate the filesystem's cache: call
 `fs.invalidate_cache()` after a change, or make it through the filesystem.
 
 ```python
@@ -357,6 +359,42 @@ for batch in S3DeleteBatch.from_paths(paths):
         print(error)  # path (code: message)
 ```
 
+`get_object_tagging()` returns the tags of an object as a dictionary, and
+`put_object_tagging()` replaces all of its tags with the given ones. Both act on the
+version ID of the path, if any, including `null`. `put_object_acl()` and
+`put_bucket_acl()` apply a canned ACL, which must be in `S3Core.OBJECT_ACLS` or
+`S3Core.BUCKET_ACLS`, respectively; another value raises `ValueError`, and an object
+ACL also applies to the version ID of the path.
+
+`replace_object_metadata()` replaces the user-defined metadata of an object, given the
+`head_object()` result of the same path, by copying the object onto itself. The copy
+rewrites the object, or creates a new version in a versioned bucket, so the path must
+not have a version ID. The copy retains the content headers (`CacheControl`,
+`ContentDisposition`, `ContentEncoding`, `ContentLanguage`, `ContentType`), `Expires`,
+`WebsiteRedirectLocation`, `StorageClass`, and the `ServerSideEncryption`,
+`SSEKMSKeyId` and `BucketKeyEnabled` of the object. Additional CopyObject parameters
+take precedence over the retained fields, and any encryption parameter replaces all of
+the retained encryption fields.
+
+```python
+path = S3Path.parse("s3://YOUR_S3_BUCKET/path/to/object")
+core.put_object_tagging(path, {**core.get_object_tagging(path), "tag2": "value2"})
+
+head = core.head_object(path)
+core.replace_object_metadata(path, head, {**head, "attr1": "value1"})
+```
+
+`generate_presigned_url()` signs a URL locally, for `get_object` unless another client
+method is given, and sends no request. Its parameters take precedence over the
+`Bucket`, `Key` and `VersionId` of the path. The `request_kwargs` of the core, such as
+`RequestPayer`, are not included in the signed parameters; pass them as parameters
+of `generate_presigned_url()` to sign them.
+
+```python
+url = core.generate_presigned_url(path, expires_in=600)
+upload_url = core.generate_presigned_url(path, "put_object", ContentType="text/csv")
+```
+
 ### Multipart writer
 
 `S3MultipartWriter` provides synchronous multipart requests and part planning without fsspec.

pyathena/filesystem/s3.py

@@ -161,34 +161,6 @@ class S3FileSystem(AbstractFileSystem):
     """
 
     DEFAULT_BLOCK_SIZE: int = 5 * 2**20  # 5MiB
-    # https://docs.aws.amazon.com/AmazonS3/latest/userguide/acl-overview.html#canned-acl
-    OBJECT_ACLS: frozenset[str] = frozenset(
-        {
-            "private",
-            "public-read",
-            "public-read-write",
-            "authenticated-read",
-            "aws-exec-read",
-            "bucket-owner-read",
-            "bucket-owner-full-control",
-        }
-    )
-    BUCKET_ACLS: frozenset[str] = frozenset(
-        {"private", "public-read", "public-read-write", "authenticated-read"}
-    )
-    # https://docs.aws.amazon.com/AmazonS3/latest/API/API_CopyObject.html
-    # The CopyObject parameters that set the encryption of the copy.
-    _SSE_COPY_PARAMS: frozenset[str] = frozenset(
-        {
-            "ServerSideEncryption",
-            "SSEKMSKeyId",
-            "SSEKMSEncryptionContext",
-            "BucketKeyEnabled",
-            "SSECustomerAlgorithm",
-            "SSECustomerKey",
-            "SSECustomerKeyMD5",
-        }
-    )
     PATTERN_PATH: Pattern[str] = S3Path.PATTERN
 
     protocol = ("s3", "s3a")
@@ -1282,8 +1254,8 @@ def mkdir(self, path: str, create_parents: bool = True, **kwargs) -> None:
                     "Set allow_bucket_creation=True on the filesystem to enable it."
                 )
             acl = kwargs.pop("acl", "")
-            if acl and acl not in self.BUCKET_ACLS:
-                raise ValueError(f"ACL not in {self.BUCKET_ACLS}.")
+            if acl and acl not in self.core.BUCKET_ACLS:
+                raise ValueError(f"ACL not in {self.core.BUCKET_ACLS}.")
             request: dict[str, Any] = {"Bucket": s3_path.bucket}
             if acl:
                 request.update({"ACL": acl})
@@ -2197,23 +2169,9 @@ def sign(self, path: str, expiration: int = 3600, **kwargs):
             ...     client_method="put_object"
             ... )
         """
-        s3_path = S3Path.parse(path)
         client_method = kwargs.pop("client_method", "get_object")
-        params = {"Bucket": s3_path.bucket, "Key": s3_path.key}
-        if s3_path.version_id:
-            params.update({"VersionId": s3_path.version_id})
-        if kwargs:
-            params.update(kwargs)
-        request = {
-            "ClientMethod": client_method,
-            "Params": params,
-            "ExpiresIn": expiration,
-        }
-
-        _logger.debug(f"Generate signed url: {s3_path.uri}")
-        return self._call(
-            self._client.generate_presigned_url,
-            **request,
+        return self.core.generate_presigned_url(
+            S3Path.parse(path), client_method, expiration, **kwargs
         )
 
     def metadata(self, path: str, **kwargs) -> S3Metadata:
@@ -2295,43 +2253,7 @@ def setxattr(self, path: str, copy_kwargs: dict[str, Any] | None = None, **kw_ar
                 metadata.pop(k, None)
             else:
                 metadata[k] = v
-
-        # With the REPLACE directive, S3 does not copy what the request
-        # omits: the system-defined metadata is dropped, and the copy is
-        # written as STANDARD with the default encryption of the bucket.
-        kept: dict[str, Any] = {
-            "CacheControl": head.cache_control,
-            "ContentDisposition": head.content_disposition,
-            "ContentEncoding": head.content_encoding,
-            "ContentLanguage": head.content_language,
-            "ContentType": head.content_type,
-            "Expires": head.expires,
-            "WebsiteRedirectLocation": head.website_redirect_location,
-            "StorageClass": head.storage_class,
-        }
-        copy_kwargs = copy_kwargs if copy_kwargs else {}
-        if not self._SSE_COPY_PARAMS.intersection(copy_kwargs):
-            kept.update(
-                {
-                    "ServerSideEncryption": head.server_side_encryption,
-                    "SSEKMSKeyId": head.sse_kms_key_id,
-                    "BucketKeyEnabled": head.bucket_key_enabled,
-                }
-            )
-
-        _logger.debug(f"Set object metadata: {s3_path.uri}")
-        self._call(
-            self._client.copy_object,
-            CopySource={"Bucket": s3_path.bucket, "Key": s3_path.key},
-            Bucket=s3_path.bucket,
-            Key=s3_path.key,
-            Metadata=metadata,
-            MetadataDirective="REPLACE",
-            **{
-                **{k: v for k, v in kept.items() if v is not None},
-                **copy_kwargs,
-            },
-        )
+        self.core.replace_object_metadata(s3_path, head, metadata, **(copy_kwargs or {}))
         self.invalidate_cache(path)
 
     def get_tags(self, path: str) -> dict[str, str]:
@@ -2346,16 +2268,7 @@ def get_tags(self, path: str) -> dict[str, str]:
         s3_path = S3Path.parse(path)
         if not s3_path.key:
             raise ValueError("Cannot get tags of a bucket.")
-        request: dict[str, Any] = {"Bucket": s3_path.bucket, "Key": s3_path.key}
-        if s3_path.version_id:
-            request.update({"VersionId": s3_path.version_id})
-
-        _logger.debug(f"Get object tagging: {s3_path.uri}")
-        response = self._call(
-            self._client.get_object_tagging,
-            **request,
-        )
-        return {v["Key"]: v["Value"] for v in response["TagSet"]}
+        return self.core.get_object_tagging(s3_path)
 
     def put_tags(self, path: str, tags: dict[str, str], mode: str = "o") -> None:
         """Set the tags for the given existing key.
@@ -2377,26 +2290,12 @@ def put_tags(self, path: str, tags: dict[str, str], mode: str = "o") -> None:
         if not s3_path.key:
             raise ValueError("Cannot put tags of a bucket.")
         if mode == "m":
-            existing_tags = self.get_tags(path)
-            existing_tags.update(tags)
-            new_tags = [{"Key": k, "Value": v} for k, v in existing_tags.items()]
+            new_tags = {**self.core.get_object_tagging(s3_path), **tags}
         elif mode == "o":
-            new_tags = [{"Key": k, "Value": v} for k, v in tags.items()]
+            new_tags = tags
         else:
             raise ValueError(f"Mode must be {{'o', 'm'}}, not {mode}.")
-        request: dict[str, Any] = {
-            "Bucket": s3_path.bucket,
-            "Key": s3_path.key,
-            "Tagging": {"TagSet": new_tags},
-        }
-        if s3_path.version_id:
-            request.update({"VersionId": s3_path.version_id})
-
-        _logger.debug(f"Put object tagging: {s3_path.uri}")
-        self._call(
-            self._client.put_object_tagging,
-            **request,
-        )
+        self.core.put_object_tagging(s3_path, new_tags)
 
     def chmod(self, path: str, acl: str, recursive: bool = False, **kwargs) -> None:
         """Set the Access Control on a bucket/key.
@@ -2414,10 +2313,10 @@ def chmod(self, path: str, acl: str, recursive: bool = False, **kwargs) -> None:
         s3_path = S3Path.parse(path)
         # Validate before any ACL is applied so that a recursive call cannot
         # partially apply object ACLs and then fail on the bucket ACL.
-        if not s3_path.key and acl not in self.BUCKET_ACLS:
-            raise ValueError(f"ACL not in {self.BUCKET_ACLS}.")
-        if s3_path.key and acl not in self.OBJECT_ACLS:
-            raise ValueError(f"ACL not in {self.OBJECT_ACLS}.")
+        if not s3_path.key and acl not in self.core.BUCKET_ACLS:
+            raise ValueError(f"ACL not in {self.core.BUCKET_ACLS}.")
+        if s3_path.key and acl not in self.core.OBJECT_ACLS:
+            raise ValueError(f"ACL not in {self.core.OBJECT_ACLS}.")
         if recursive:
             with self._create_executor(max_workers=self.max_workers) as executor:
                 futures = [
@@ -2431,24 +2330,9 @@ def chmod(self, path: str, acl: str, recursive: bool = False, **kwargs) -> None:
                 # below it have ACLs.
                 return
         if s3_path.key:
-            request: dict[str, Any] = {"Bucket": s3_path.bucket, "Key": s3_path.key, "ACL": acl}
-            if s3_path.version_id:
-                request.update({"VersionId": s3_path.version_id})
-
-            _logger.debug(f"Put object acl: {s3_path.uri}")
-            self._call(
-                self._client.put_object_acl,
-                **request,
-                **kwargs,
-            )
+            self.core.put_object_acl(s3_path, acl, **kwargs)
         else:
-            _logger.debug(f"Put bucket acl: {s3_path.uri}")
-            self._call(
-                self._client.put_bucket_acl,
-                Bucket=s3_path.bucket,
-                ACL=acl,
-                **kwargs,
-            )
+            self.core.put_bucket_acl(s3_path.bucket, acl, **kwargs)
 
     def list_multipart_uploads(self, path: str) -> list[S3MultipartUpload]:
         """List in-progress (incomplete) multipart uploads in a bucket.