Method mkdir

pyathena/filesystem/s3.py:837–908

Create an S3 bucket. S3 has no real directories below the bucket level; creating a key prefix requires no operation. This method creates the bucket when the path points at a bucket (or when ``create_parents`` is True and the bucket does not exist yet), and does nothi

(self, path: str, create_parents: bool = True, **kwargs)

                f.result()

    def mkdir(self, path: str, create_parents: bool = True, **kwargs) -> None:
        """Create an S3 bucket.

        S3 has no real directories below the bucket level; creating a key
        prefix requires no operation. This method creates the bucket when the
        path points at a bucket (or when ``create_parents`` is True and the
        bucket does not exist yet), and does nothing for key prefixes under
        an existing bucket.

        Bucket lifecycle operations are disabled by default because they are
        infrastructure-level changes; pass ``allow_bucket_creation=True`` to
        the filesystem constructor to enable bucket creation.

        Args:
            path: S3 path (e.g., "s3://bucket" or "s3://bucket/prefix").
            create_parents: If True, create the bucket when it does not exist,
                even if the path contains a key prefix.
            **kwargs: Additional arguments including:
                acl: Canned ACL to apply to the bucket.
                region_name: Region to create the bucket in. Defaults to the
                    client's region.

        Raises:
            FileExistsError: If the path is a bucket that already exists.
            FileNotFoundError: If the bucket does not exist and
                ``create_parents`` is False.
            PermissionError: If the bucket would be created but bucket
                creation is not enabled on this filesystem instance.
            ValueError: If the ACL is invalid or the path is empty.
        """
        path = self._strip_protocol(path).rstrip("/")
        if not path:
            raise ValueError("Cannot create the root directory.")
        bucket, key, _ = self.parse_path(path)
        if self.exists(bucket):
            if not key:
                # Requested to create a bucket, but the bucket already exists.
                raise FileExistsError(bucket)
            # Do nothing as the bucket already exists.
        elif not key or create_parents:
            if not self.allow_bucket_creation:
                raise PermissionError(
                    "Bucket creation is disabled. "
                    "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}.")
            request: dict[str, Any] = {"Bucket": bucket}
            if acl:
                request.update({"ACL": acl})
            region_name = kwargs.pop("region_name", None) or self._client.meta.region_name
            if region_name and region_name != "us-east-1":
                # us-east-1 does not accept a location constraint.
                request.update({"CreateBucketConfiguration": {"LocationConstraint": region_name}})

            _logger.debug(f"Create bucket: s3://{bucket}")
            try: