Method setxattr

pyathena/filesystem/s3.py:1459–1507

Set the user-defined metadata of the path. S3 does not allow updating the metadata of an existing object in place, so the object is copied onto itself with the REPLACE metadata directive. Note that this rewrites the object and updates its last-modified time.

(self, path: str, copy_kwargs: dict[str, Any] | None = None, **kw_args)

        return self.metadata(path, **kwargs).get(attr_name)

    def setxattr(self, path: str, copy_kwargs: dict[str, Any] | None = None, **kw_args) -> None:
        """Set the user-defined metadata of the path.

        S3 does not allow updating the metadata of an existing object in
        place, so the object is copied onto itself with the REPLACE metadata
        directive. Note that this rewrites the object and updates its
        last-modified time.

        Args:
            path: S3 path (s3://bucket/key) to set metadata for.
            copy_kwargs: Additional parameters to use for the underlying
                CopyObject API call.
            **kw_args: Key-value pairs to set, where the values must be
                strings. The keys are used as-is; names that are not valid
                Python identifiers (e.g., containing hyphens) can be passed
                by unpacking a dictionary. Does not alter existing fields,
                unless the field appears here - if the value is None, delete
                the field.

        Example:
            >>> fs = S3FileSystem()
            >>> fs.setxattr("s3://bucket/key", attribute1="value1")
            >>> fs.setxattr("s3://bucket/key", **{"attribute-2": "value2"})
        """
        bucket, key, version_id = self.parse_path(path)
        if not key:
            raise ValueError("Cannot set metadata of a bucket.")
        metadata = dict(self.metadata(path))
        for k, v in kw_args.items():
            if v is None:
                metadata.pop(k, None)
            else:
                metadata[k] = v

        copy_source: dict[str, Any] = {"Bucket": bucket, "Key": key}
        if version_id:
            copy_source.update({"VersionId": version_id})

        _logger.debug(f"Set object metadata: s3://{bucket}/{key}?versionId={version_id}")
        self._call(
            self._client.copy_object,
            CopySource=copy_source,
            Bucket=bucket,
            Key=key,
            Metadata=metadata,
            MetadataDirective="REPLACE",
            **(copy_kwargs if copy_kwargs else {}),
        )
        self.invalidate_cache(path)

    def get_tags(self, path: str) -> dict[str, str]:
        """Retrieve the tag key/values for the given path.