Source code for s3pathlib.core.mutate

# -*- coding: utf-8 -*-

"""
S3Path object mutation implementation.
"""

import typing as T

from .. import exc

if T.TYPE_CHECKING:  # pragma: no cover
    from .s3path import S3Path




class MutateAPIMixin:
    """
    A mixin class that implements the S3Path object mutation.
    """



    def copy(self: "S3Path") -> "S3Path":
        """
        Create a copy of S3Path object that logically equals to this one,
        but is actually different identity in memory. Also, the cache data
        are cleared.

        Example::

            >>> p1 = S3Path("bucket", "folder", "file.txt")
            >>> p2 = p1.copy()
            >>> p1 == p2
            True
            >>> p1 is p2
            False

        .. versionadded:: 1.0.1
        """
        return self._from_parsed_parts(
            bucket=self._bucket,
            parts=list(self._parts),
            is_dir=self._is_dir,
        )




    def change(
        self: "S3Path",
        new_bucket: str = None,
        new_abspath: str = None,
        new_dirpath: str = None,
        new_dirname: str = None,
        new_basename: str = None,
        new_fname: str = None,
        new_ext: str = None,
    ) -> "S3Path":
        """
        Create a new S3Path by replacing part of the attributes. If no argument
        is given, it behaves like :meth:`copy`.

        Example:

            >>> s3path = S3Path("bucket", "folder", "file.txt")
            >>> s3path.change(new_bucket="new_bucket")
            S3Path('s3://new_bucket/folder/file.txt')

            >>> s3path = S3Path("bucket", "folder", "file.txt")
            >>> s3path.change(new_basename="data.json")
            S3Path('s3://bucket/folder/data.json')

            >>> s3path = S3Path("bucket", "folder", "file.txt")
            >>> s3path.change(new_fname="log")
            S3Path('s3://bucket/folder/log.txt')

        :param new_bucket: The new bucket name
        :param new_abspath:
        :param new_dirpath:
        :param new_dirname:
        :param new_basename:
        :param new_fname:
        :param new_ext:

        .. versionadded:: 1.0.2
        """
        if new_bucket is None:
            new_bucket = self.bucket

        if new_abspath is not None:
            exc.ensure_all_none(
                new_dirpath=new_dirpath,
                new_dirname=new_dirname,
                new_basename=new_basename,
                new_fname=new_fname,
                new_ext=new_ext,
            )
            p = self._from_parts([self.bucket, new_abspath])
            return p

        if (new_dirpath is None) and (new_dirname is not None):
            dir_parts = self.parent.parent._parts + [new_dirname]
        elif (new_dirpath is not None) and (new_dirname is None):
            dir_parts = [
                new_dirpath,
            ]
        elif (new_dirpath is None) and (new_dirname is None):
            dir_parts = self.parent._parts
        else:
            raise ValueError("Cannot having both 'new_dirpath' and 'new_dirname'!")

        if new_basename is None:
            if new_fname is None:
                new_fname = self.fname
            if new_ext is None:
                new_ext = self.ext
            new_basename = new_fname + new_ext
        else:
            if (new_fname is not None) or (new_ext is not None):
                raise ValueError(
                    "Cannot having both "
                    "'new_basename' / 'new_fname', "
                    "or 'new_basename' / 'new_ext'!"
                )
        if new_bucket is None:
            p = self._from_parts(
                [
                    "dummy-bucket",
                ]
                + dir_parts
                + [
                    new_basename,
                ]
            )
            p._bucket = None
        else:
            p = self._from_parts(
                [
                    new_bucket,
                ]
                + dir_parts
                + [
                    new_basename,
                ]
            )
        return p




    def to_dir(self: "S3Path") -> "S3Path":
        """
        Convert the S3Path to a directory. If the S3Path is a file, then append
        a "/" at the end. If the S3Path is already a directory, then do nothing.

        Example:

            >>> S3Path.from_s3_uri("s3://bucket/folder").to_dir()
            S3Path('s3://bucket/folder/')
        """
        if self.is_dir():
            return self.copy()
        elif self.is_file():
            return self.joinpath("/")
        else:
            raise ValueError("only concrete file or folder S3Path can do .to_dir()")




    def to_file(self: "S3Path") -> "S3Path":
        """
        Convert the S3Path to a file. If the S3Path is a directory, then strip
        out the last "/". If the S3Path is already a file, then do nothing.

        Example:

            >>> S3Path.from_s3_uri("s3://bucket/file/").to_dir()
            S3Path('s3://bucket/file/')
        """
        if self.is_file():
            return self.copy()
        elif self.is_dir():
            p = self.copy()
            p._is_dir = False
            return p
        else:
            raise ValueError("only concrete file or folder S3Path can do .to_file()")