    async def _pipe_file(
        self,
        path,
        data,
        chunksize=50 * 2**20,
        max_concurrency=None,
        mode="overwrite",
        **kwargs,
    ):
        """
        mode=="create", exclusive write, is only known to work on AWS S3, and
        requires botocore>1.35.20
        """
        bucket, key, _ = self.split_path(path)
        concurrency = max_concurrency or self.max_concurrency
        size = len(data)
        if mode == "create":
            match = {"IfNoneMatch": "*"}
        else:
            match = {}

        # 5 GB is the limit for an S3 PUT
        if size < min(5 * 2**30, 2 * chunksize):
            out = await self._call_s3(
                "put_object", Bucket=bucket, Key=key, Body=data, **kwargs, **match
            )
            self.invalidate_cache(path)
            return out
        else:

            mpu = await self._call_s3(
                "create_multipart_upload", Bucket=bucket, Key=key, **kwargs
            )
            ranges = list(range(0, len(data), chunksize))
            inds = list(range(0, len(ranges), concurrency)) + [len(ranges)]
            parts = []
            try:
                for start, stop in zip(inds[:-1], inds[1:]):
                    out = await asyncio.gather(
                        *[
                            self._call_s3(
                                "upload_part",
                                Bucket=bucket,
                                PartNumber=i + 1,
                                UploadId=mpu["UploadId"],
                                Body=data[ranges[i] : ranges[i] + chunksize],
                                Key=key,
                            )
                            for i in range(start, stop)
                        ]
                    )
                    parts.extend(
                        {"PartNumber": i + 1, "ETag": o["ETag"]}
                        for i, o in zip(range(start, stop), out)
                    )
                await self._call_s3(
                    "complete_multipart_upload",
                    Bucket=bucket,
                    Key=key,
                    UploadId=mpu["UploadId"],
                    MultipartUpload={"Parts": parts},
                    **match,
                )
                self.invalidate_cache(path)
            except Exception:
                await self._abort_mpu(bucket, key, mpu["UploadId"])
                raise
mode=="create", exclusive write, is only known to work on AWS S3, and
requires botocore>1.35.20
s3fs/core.py