Method pipe_file

pyathena/filesystem/s3.py:1138–1193

Write bytes into the path. Writes data up to the block size with a single PutObject request instead of the inherited ``open()`` + ``write()`` path. Larger data and writes inside an fsspec transaction go through the buffered path, which uploads the data as a parallel

(
        self, path: str, value: bytes | bytearray | memoryview, mode: str = "overwrite", **kwargs
    )

            )

    def pipe_file(
        self, path: str, value: bytes | bytearray | memoryview, mode: str = "overwrite", **kwargs
    ) -> None:
        """Write bytes into the path.

        Writes data up to the block size with a single PutObject request
        instead of the inherited ``open()`` + ``write()`` path. Larger data
        and writes inside an fsspec transaction go through the buffered
        path, which uploads the data as a parallel multipart upload and
        keeps the deferred-commit semantics of transactions.

        Args:
            path: S3 path (s3://bucket/key) to write to.
            value: The bytes to write.
            mode: "overwrite" (default) or "create". With "create", raise
                FileExistsError when the object already exists.
            **kwargs: Additional parameters passed to the PutObject API
                (e.g., ContentType, StorageClass) on the single-request
                path. The ``block_size``, ``max_worker``, and
                ``s3_additional_kwargs`` parameters of the ``open()`` path
                are also accepted.

        Raises:
            FileExistsError: If the mode is "create" and the path already
                exists.
            ValueError: If the path does not contain a key or specifies a
                version.
        """
        block_size = kwargs.get("block_size") or self.default_block_size
        if self._intrans or len(value) > min(block_size, self.MULTIPART_UPLOAD_MAX_PART_SIZE):
            # Defer to the buffered open() path, which keeps the
            # deferred-commit semantics of fsspec transactions and uploads
            # large data as a parallel multipart upload.
            super().pipe_file(path, value, mode=mode, **kwargs)
            return
        bucket, key, version_id = self.parse_path(path)
        if version_id:
            raise ValueError("Cannot write to the file with the version specified.")
        if not key:
            raise ValueError("Cannot write to a bucket.")
        if mode == "create" and self.exists(path):
            raise FileExistsError(path)
        if not isinstance(value, bytes):
            # Accept bytes-like values (bytearray, memoryview) as the
            # buffered path does.
            value = bytes(value)

        kwargs.pop("block_size", None)
        kwargs.pop("max_worker", None)
        request_kwargs = {
            **self.s3_additional_kwargs,
            **kwargs.pop("s3_additional_kwargs", {}),
            **kwargs,
        }
        self._put_object(bucket=bucket, key=key, body=value, **request_kwargs)
        self.invalidate_cache(path)

    def _finish_multipart_upload(