cloudpathlib.local

This module implements "Local" classes that mimic their associated cloudpathlib non-local counterparts but use the local filesystem in place of cloud storage. They can be used as drop-in replacements, with the intent that you can use them as mock or monkepatch substitutes in your tests. See "Testing code that uses cloudpathlib" for usage examples.

Modules

implementations special

Modules

azure

Attributes

local_azure_blob_implementation

Replacement for "azure" CloudImplementation meta object in cloudpathlib.implementation_registry

Classes

LocalAzureBlobClient (LocalClient)

Replacement for AzureBlobClient that uses the local file system. Intended as a monkeypatch substitute when writing tests.

Source code in cloudpathlib/local/implementations/azure.py

class LocalAzureBlobClient(LocalClient):
    """Replacement for AzureBlobClient that uses the local file system. Intended as a monkeypatch
    substitute when writing tests.
    """

    _cloud_meta = local_azure_blob_implementation

    def __init__(self, *args, **kwargs):
        cred_opts = [
            kwargs.get("blob_service_client", None),
            kwargs.get("connection_string", None),
            kwargs.get("account_url", None),
            os.getenv("AZURE_STORAGE_CONNECTION_STRING", None),
        ]
        super().__init__(*args, **kwargs)

        if all(opt is None for opt in cred_opts):
            raise MissingCredentialsError(
                "AzureBlobClient does not support anonymous instantiation. "
                "Credentials are required; see docs for options."
            )

AzureBlobPath(self, cloud_path: Union[str, ~BoundedCloudPath]) -> ~BoundedCloudPath

Source code in cloudpathlib/local/implementations/azure.py

def CloudPath(self, cloud_path: Union[str, BoundedCloudPath]) -> BoundedCloudPath:
    return self._cloud_meta.path_class(cloud_path=cloud_path, client=self)  # type: ignore

__init__(self, *args, **kwargs) special

Source code in cloudpathlib/local/implementations/azure.py

def __init__(self, *args, **kwargs):
    cred_opts = [
        kwargs.get("blob_service_client", None),
        kwargs.get("connection_string", None),
        kwargs.get("account_url", None),
        os.getenv("AZURE_STORAGE_CONNECTION_STRING", None),
    ]
    super().__init__(*args, **kwargs)

    if all(opt is None for opt in cred_opts):
        raise MissingCredentialsError(
            "AzureBlobClient does not support anonymous instantiation. "
            "Credentials are required; see docs for options."
        )

LocalAzureBlobPath (LocalPath)

Replacement for AzureBlobPath that uses the local file system. Intended as a monkeypatch substitute when writing tests.

Attributes

blob: str property readonly

cloud_prefix: str

container: str property readonly

drive: str property readonly

The drive prefix (letter or UNC path), if any. (Docstring copied from pathlib.Path)

etag property readonly

md5: str property readonly

Methods

mkdir(self, parents = False, exist_ok = False)

Create a new directory at this given path. (Docstring copied from pathlib.Path)

Source code in cloudpathlib/local/implementations/azure.py

def mkdir(self, parents=False, exist_ok=False):
    # not possible to make empty directory on blob storage
    pass

gs

Attributes

local_gs_implementation

Replacement for "gs" CloudImplementation meta object in cloudpathlib.implementation_registry

Classes

LocalGSClient (LocalClient)

Replacement for GSClient that uses the local file system. Intended as a monkeypatch substitute when writing tests.

Source code in cloudpathlib/local/implementations/gs.py

class LocalGSClient(LocalClient):
    """Replacement for GSClient that uses the local file system. Intended as a monkeypatch
    substitute when writing tests.
    """

    _cloud_meta = local_gs_implementation

GSPath(self, cloud_path: Union[str, ~BoundedCloudPath]) -> ~BoundedCloudPath

Source code in cloudpathlib/local/implementations/gs.py

def CloudPath(self, cloud_path: Union[str, BoundedCloudPath]) -> BoundedCloudPath:
    return self._cloud_meta.path_class(cloud_path=cloud_path, client=self)  # type: ignore

LocalGSPath (LocalPath)

Replacement for GSPath that uses the local file system. Intended as a monkeypatch substitute when writing tests.

Attributes

blob: str property readonly

bucket: str property readonly

cloud_prefix: str

drive: str property readonly

The drive prefix (letter or UNC path), if any. (Docstring copied from pathlib.Path)

etag property readonly

Methods

mkdir(self, parents = False, exist_ok = False)

Create a new directory at this given path. (Docstring copied from pathlib.Path)

Source code in cloudpathlib/local/implementations/gs.py

def mkdir(self, parents=False, exist_ok=False):
    # not possible to make empty directory on gs
    pass

s3

Attributes

local_s3_implementation

Replacement for "s3" CloudImplementation meta object in cloudpathlib.implementation_registry

Classes

LocalS3Client (LocalClient)

Replacement for S3Client that uses the local file system. Intended as a monkeypatch substitute when writing tests.

Source code in cloudpathlib/local/implementations/s3.py

class LocalS3Client(LocalClient):
    """Replacement for S3Client that uses the local file system. Intended as a monkeypatch
    substitute when writing tests.
    """

    _cloud_meta = local_s3_implementation

S3Path(self, cloud_path: Union[str, ~BoundedCloudPath]) -> ~BoundedCloudPath

Source code in cloudpathlib/local/implementations/s3.py

def CloudPath(self, cloud_path: Union[str, BoundedCloudPath]) -> BoundedCloudPath:
    return self._cloud_meta.path_class(cloud_path=cloud_path, client=self)  # type: ignore

LocalS3Path (LocalPath)

Replacement for S3Path that uses the local file system. Intended as a monkeypatch substitute when writing tests.

Attributes

bucket: str property readonly

cloud_prefix: str

drive: str property readonly

The drive prefix (letter or UNC path), if any. (Docstring copied from pathlib.Path)

etag property readonly

key: str property readonly

Methods

mkdir(self, parents = False, exist_ok = False)

Create a new directory at this given path. (Docstring copied from pathlib.Path)

Source code in cloudpathlib/local/implementations/s3.py

def mkdir(self, parents=False, exist_ok=False):
    # not possible to make empty directory on s3
    pass

localclient

Classes

LocalClient (Client)

Abstract client for accessing objects the local filesystem. Subclasses are as a monkeypatch substitutes for normal Client subclasses when writing tests.

Source code in cloudpathlib/local/localclient.py

class LocalClient(Client):
    """Abstract client for accessing objects the local filesystem. Subclasses are as a monkeypatch
    substitutes for normal Client subclasses when writing tests."""

    # Class-level variable to tracks the default storage directory for this client class
    # that is used if a client is instantiated without a directory being explicitly provided
    _default_storage_temp_dir: ClassVar[Optional[TemporaryDirectory]] = None

    # Instance-level variable that tracks the local storage directory for this client
    _local_storage_dir: Optional[Union[str, os.PathLike]]

    def __init__(
        self,
        *args,
        local_storage_dir: Optional[Union[str, os.PathLike]] = None,
        file_cache_mode: Optional[Union[str, FileCacheMode]] = None,
        local_cache_dir: Optional[Union[str, os.PathLike]] = None,
        content_type_method: Optional[Callable] = mimetypes.guess_type,
        **kwargs,
    ):
        self._local_storage_dir = local_storage_dir

        super().__init__(
            local_cache_dir=local_cache_dir,
            content_type_method=content_type_method,
            file_cache_mode=file_cache_mode,
        )

    @classmethod
    def get_default_storage_dir(cls) -> Path:
        """Return the default storage directory for this client class. This is used if a client
        is instantiated without a storage directory being explicitly provided. In this usage,
        "storage" refers to the local storage that simulates the cloud.
        """
        if cls._default_storage_temp_dir is None:
            cls._default_storage_temp_dir = TemporaryDirectory()
            _temp_dirs_to_clean.append(cls._default_storage_temp_dir)
        return Path(cls._default_storage_temp_dir.name)

    @classmethod
    def reset_default_storage_dir(cls) -> Path:
        """Reset the default storage directly. This tears down and recreates the directory used by
        default for this client class when instantiating a client without explicitly providing
        a storage directory. In this usage, "storage" refers to the local storage that simulates
        the cloud.
        """
        cls._default_storage_temp_dir = None
        return cls.get_default_storage_dir()

    @property
    def local_storage_dir(self) -> Path:
        """The local directory where files are stored for this client. This storage directory is
        the one that simulates the cloud. If no storage directory was provided on instantiating the
        client, the default storage directory for this client class is used.
        """
        if self._local_storage_dir is None:
            # No explicit local storage was provided on instantiating the client.
            # Use the default storage directory for this class.
            return self.get_default_storage_dir()
        return Path(self._local_storage_dir)

    def _cloud_path_to_local(self, cloud_path: "LocalPath") -> Path:
        return self.local_storage_dir / cloud_path._no_prefix

    def _local_to_cloud_path(self, local_path: Union[str, os.PathLike]) -> "LocalPath":
        local_path = Path(local_path)
        cloud_prefix = self._cloud_meta.path_class.cloud_prefix
        return self.CloudPath(
            f"{cloud_prefix}{PurePosixPath(local_path.relative_to(self.local_storage_dir))}"
        )

    def _download_file(self, cloud_path: "LocalPath", local_path: Union[str, os.PathLike]) -> Path:
        local_path = Path(local_path)
        local_path.parent.mkdir(exist_ok=True, parents=True)

        try:
            shutil.copyfile(self._cloud_path_to_local(cloud_path), local_path)
        except FileNotFoundError:
            # erroneous FileNotFoundError appears in tests sometimes; patiently insist on the parent directory existing
            sleep(1.0)
            local_path.parent.mkdir(exist_ok=True, parents=True)
            sleep(1.0)

            shutil.copyfile(self._cloud_path_to_local(cloud_path), local_path)

        return local_path

    def _exists(self, cloud_path: "LocalPath") -> bool:
        return self._cloud_path_to_local(cloud_path).exists()

    def _is_dir(self, cloud_path: "LocalPath") -> bool:
        return self._cloud_path_to_local(cloud_path).is_dir()

    def _is_file(self, cloud_path: "LocalPath") -> bool:
        return self._cloud_path_to_local(cloud_path).is_file()

    def _list_dir(
        self, cloud_path: "LocalPath", recursive=False
    ) -> Iterable[Tuple["LocalPath", bool]]:
        pattern = "**/*" if recursive else "*"
        for obj in self._cloud_path_to_local(cloud_path).glob(pattern):
            yield (self._local_to_cloud_path(obj), obj.is_dir())

    def _md5(self, cloud_path: "LocalPath") -> str:
        return md5(self._cloud_path_to_local(cloud_path).read_bytes()).hexdigest()

    def _move_file(
        self, src: "LocalPath", dst: "LocalPath", remove_src: bool = True
    ) -> "LocalPath":
        self._cloud_path_to_local(dst).parent.mkdir(exist_ok=True, parents=True)

        if remove_src:
            self._cloud_path_to_local(src).replace(self._cloud_path_to_local(dst))
        else:
            shutil.copy(self._cloud_path_to_local(src), self._cloud_path_to_local(dst))
        return dst

    def _remove(self, cloud_path: "LocalPath", missing_ok: bool = True) -> None:
        local_storage_path = self._cloud_path_to_local(cloud_path)
        if not missing_ok and not local_storage_path.exists():
            raise FileNotFoundError(f"File does not exist: {cloud_path}")

        if local_storage_path.is_file():
            local_storage_path.unlink()
        elif local_storage_path.is_dir():
            shutil.rmtree(local_storage_path)

    def _stat(self, cloud_path: "LocalPath") -> os.stat_result:
        stat_result = self._cloud_path_to_local(cloud_path).stat()

        return os.stat_result(
            (  # type: ignore
                None,  # type: ignore # mode
                None,  # ino
                cloud_path.cloud_prefix,  # dev,
                None,  # nlink,
                None,  # uid,
                None,  # gid,
                stat_result.st_size,  # size,
                None,  # atime,
                stat_result.st_mtime,  # mtime,
                None,  # ctime,
            )
        )

    def _touch(self, cloud_path: "LocalPath", exist_ok: bool = True) -> None:
        local_storage_path = self._cloud_path_to_local(cloud_path)
        if local_storage_path.exists() and not exist_ok:
            raise FileExistsError(f"File exists: {cloud_path}")
        local_storage_path.parent.mkdir(exist_ok=True, parents=True)
        local_storage_path.touch()

    def _upload_file(
        self, local_path: Union[str, os.PathLike], cloud_path: "LocalPath"
    ) -> "LocalPath":
        dst = self._cloud_path_to_local(cloud_path)
        dst.parent.mkdir(exist_ok=True, parents=True)
        shutil.copy(local_path, dst)
        return cloud_path

    def _get_metadata(self, cloud_path: "LocalPath") -> Dict:
        # content_type is the only metadata we test currently
        if self.content_type_method is None:
            content_type_method = lambda x: (None, None)
        else:
            content_type_method = self.content_type_method

        return {
            "content_type": content_type_method(str(self._cloud_path_to_local(cloud_path)))[0],
        }

    def _get_public_url(self, cloud_path: "LocalPath") -> str:
        return cloud_path.as_uri()

    def _generate_presigned_url(
        self, cloud_path: "LocalPath", expire_seconds: int = 60 * 60
    ) -> str:
        raise NotImplementedError("Cannot generate a presigned URL for a local path.")

Attributes

local_storage_dir: Path property readonly

The local directory where files are stored for this client. This storage directory is the one that simulates the cloud. If no storage directory was provided on instantiating the client, the default storage directory for this client class is used.

Methods

__init__(self, *args, *, local_storage_dir: Union[str, os.PathLike] = None, file_cache_mode: Union[str, cloudpathlib.enums.FileCacheMode] = None, local_cache_dir: Union[str, os.PathLike] = None, content_type_method: Optional[Callable] = <function guess_type at 0x7f9a1cd285e0>, **kwargs) special

Source code in cloudpathlib/local/localclient.py

def __init__(
    self,
    *args,
    local_storage_dir: Optional[Union[str, os.PathLike]] = None,
    file_cache_mode: Optional[Union[str, FileCacheMode]] = None,
    local_cache_dir: Optional[Union[str, os.PathLike]] = None,
    content_type_method: Optional[Callable] = mimetypes.guess_type,
    **kwargs,
):
    self._local_storage_dir = local_storage_dir

    super().__init__(
        local_cache_dir=local_cache_dir,
        content_type_method=content_type_method,
        file_cache_mode=file_cache_mode,
    )

get_default_storage_dir() -> Path classmethod

Return the default storage directory for this client class. This is used if a client is instantiated without a storage directory being explicitly provided. In this usage, "storage" refers to the local storage that simulates the cloud.

Source code in cloudpathlib/local/localclient.py

@classmethod
def get_default_storage_dir(cls) -> Path:
    """Return the default storage directory for this client class. This is used if a client
    is instantiated without a storage directory being explicitly provided. In this usage,
    "storage" refers to the local storage that simulates the cloud.
    """
    if cls._default_storage_temp_dir is None:
        cls._default_storage_temp_dir = TemporaryDirectory()
        _temp_dirs_to_clean.append(cls._default_storage_temp_dir)
    return Path(cls._default_storage_temp_dir.name)

reset_default_storage_dir() -> Path classmethod

Reset the default storage directly. This tears down and recreates the directory used by default for this client class when instantiating a client without explicitly providing a storage directory. In this usage, "storage" refers to the local storage that simulates the cloud.

Source code in cloudpathlib/local/localclient.py

@classmethod
def reset_default_storage_dir(cls) -> Path:
    """Reset the default storage directly. This tears down and recreates the directory used by
    default for this client class when instantiating a client without explicitly providing
    a storage directory. In this usage, "storage" refers to the local storage that simulates
    the cloud.
    """
    cls._default_storage_temp_dir = None
    return cls.get_default_storage_dir()

clean_temp_dirs()

Source code in cloudpathlib/local/localclient.py

@atexit.register
def clean_temp_dirs():
    for temp_dir in _temp_dirs_to_clean:
        temp_dir.cleanup()

localpath

Classes

LocalPath (CloudPath)

Abstract CloudPath for accessing objects the local filesystem. Subclasses are as a monkeypatch substitutes for normal CloudPath subclasses when writing tests.

Source code in cloudpathlib/local/localpath.py

class LocalPath(CloudPath):
    """Abstract CloudPath for accessing objects the local filesystem. Subclasses are as a
    monkeypatch substitutes for normal CloudPath subclasses when writing tests."""

    client: "LocalClient"

    def is_dir(self) -> bool:
        return self.client._is_dir(self)

    def is_file(self) -> bool:
        return self.client._is_file(self)

    def stat(self):
        try:
            meta = self.client._stat(self)
        except FileNotFoundError:
            raise NoStatError(
                f"No stats available for {self}; it may be a directory or not exist."
            )
        return meta

    def touch(self, exist_ok: bool = True):
        self.client._touch(self, exist_ok)

Methods

is_dir(self) -> bool

Whether this path is a directory. (Docstring copied from pathlib.Path)

Source code in cloudpathlib/local/localpath.py

def is_dir(self) -> bool:
    return self.client._is_dir(self)

is_file(self) -> bool

Whether this path is a regular file (also True for symlinks pointing to regular files). (Docstring copied from pathlib.Path)

Source code in cloudpathlib/local/localpath.py

def is_file(self) -> bool:
    return self.client._is_file(self)

stat(self)

Return the result of the stat() system call on this path, like os.stat() does. (Docstring copied from pathlib.Path)

Source code in cloudpathlib/local/localpath.py

def stat(self):
    try:
        meta = self.client._stat(self)
    except FileNotFoundError:
        raise NoStatError(
            f"No stats available for {self}; it may be a directory or not exist."
        )
    return meta

touch(self, exist_ok: bool = True)

Create this file with the given access mode, if it doesn't exist. (Docstring copied from pathlib.Path)

Source code in cloudpathlib/local/localpath.py

def touch(self, exist_ok: bool = True):
    self.client._touch(self, exist_ok)

Attributes

Classes

Replacement for AzureBlobClient that uses the local file system. Intended as a monkeypatch substitute when writing tests.

Source code in cloudpathlib/local/implementations/azure.py

class LocalAzureBlobClient(LocalClient):
    """Replacement for AzureBlobClient that uses the local file system. Intended as a monkeypatch
    substitute when writing tests.
    """

    _cloud_meta = local_azure_blob_implementation

    def __init__(self, *args, **kwargs):
        cred_opts = [
            kwargs.get("blob_service_client", None),
            kwargs.get("connection_string", None),
            kwargs.get("account_url", None),
            os.getenv("AZURE_STORAGE_CONNECTION_STRING", None),
        ]
        super().__init__(*args, **kwargs)

        if all(opt is None for opt in cred_opts):
            raise MissingCredentialsError(
                "AzureBlobClient does not support anonymous instantiation. "
                "Credentials are required; see docs for options."
            )

AzureBlobPath(self, cloud_path: Union[str, ~BoundedCloudPath]) -> ~BoundedCloudPath

Source code in cloudpathlib/local/implementations/azure.py

def CloudPath(self, cloud_path: Union[str, BoundedCloudPath]) -> BoundedCloudPath:
    return self._cloud_meta.path_class(cloud_path=cloud_path, client=self)  # type: ignore

__init__(self, *args, **kwargs) special

Source code in cloudpathlib/local/implementations/azure.py

def __init__(self, *args, **kwargs):
    cred_opts = [
        kwargs.get("blob_service_client", None),
        kwargs.get("connection_string", None),
        kwargs.get("account_url", None),
        os.getenv("AZURE_STORAGE_CONNECTION_STRING", None),
    ]
    super().__init__(*args, **kwargs)

    if all(opt is None for opt in cred_opts):
        raise MissingCredentialsError(
            "AzureBlobClient does not support anonymous instantiation. "
            "Credentials are required; see docs for options."
        )

Replacement for AzureBlobPath that uses the local file system. Intended as a monkeypatch substitute when writing tests.

Attributes

blob: str property readonly

cloud_prefix: str

container: str property readonly

drive: str property readonly

The drive prefix (letter or UNC path), if any. (Docstring copied from pathlib.Path)

etag property readonly

md5: str property readonly

Methods

mkdir(self, parents = False, exist_ok = False)

Create a new directory at this given path. (Docstring copied from pathlib.Path)

Source code in cloudpathlib/local/implementations/azure.py

def mkdir(self, parents=False, exist_ok=False):
    # not possible to make empty directory on blob storage
    pass

Abstract client for accessing objects the local filesystem. Subclasses are as a monkeypatch substitutes for normal Client subclasses when writing tests.

Source code in cloudpathlib/local/localclient.py

class LocalClient(Client):
    """Abstract client for accessing objects the local filesystem. Subclasses are as a monkeypatch
    substitutes for normal Client subclasses when writing tests."""

    # Class-level variable to tracks the default storage directory for this client class
    # that is used if a client is instantiated without a directory being explicitly provided
    _default_storage_temp_dir: ClassVar[Optional[TemporaryDirectory]] = None

    # Instance-level variable that tracks the local storage directory for this client
    _local_storage_dir: Optional[Union[str, os.PathLike]]

    def __init__(
        self,
        *args,
        local_storage_dir: Optional[Union[str, os.PathLike]] = None,
        file_cache_mode: Optional[Union[str, FileCacheMode]] = None,
        local_cache_dir: Optional[Union[str, os.PathLike]] = None,
        content_type_method: Optional[Callable] = mimetypes.guess_type,
        **kwargs,
    ):
        self._local_storage_dir = local_storage_dir

        super().__init__(
            local_cache_dir=local_cache_dir,
            content_type_method=content_type_method,
            file_cache_mode=file_cache_mode,
        )

    @classmethod
    def get_default_storage_dir(cls) -> Path:
        """Return the default storage directory for this client class. This is used if a client
        is instantiated without a storage directory being explicitly provided. In this usage,
        "storage" refers to the local storage that simulates the cloud.
        """
        if cls._default_storage_temp_dir is None:
            cls._default_storage_temp_dir = TemporaryDirectory()
            _temp_dirs_to_clean.append(cls._default_storage_temp_dir)
        return Path(cls._default_storage_temp_dir.name)

    @classmethod
    def reset_default_storage_dir(cls) -> Path:
        """Reset the default storage directly. This tears down and recreates the directory used by
        default for this client class when instantiating a client without explicitly providing
        a storage directory. In this usage, "storage" refers to the local storage that simulates
        the cloud.
        """
        cls._default_storage_temp_dir = None
        return cls.get_default_storage_dir()

    @property
    def local_storage_dir(self) -> Path:
        """The local directory where files are stored for this client. This storage directory is
        the one that simulates the cloud. If no storage directory was provided on instantiating the
        client, the default storage directory for this client class is used.
        """
        if self._local_storage_dir is None:
            # No explicit local storage was provided on instantiating the client.
            # Use the default storage directory for this class.
            return self.get_default_storage_dir()
        return Path(self._local_storage_dir)

    def _cloud_path_to_local(self, cloud_path: "LocalPath") -> Path:
        return self.local_storage_dir / cloud_path._no_prefix

    def _local_to_cloud_path(self, local_path: Union[str, os.PathLike]) -> "LocalPath":
        local_path = Path(local_path)
        cloud_prefix = self._cloud_meta.path_class.cloud_prefix
        return self.CloudPath(
            f"{cloud_prefix}{PurePosixPath(local_path.relative_to(self.local_storage_dir))}"
        )

    def _download_file(self, cloud_path: "LocalPath", local_path: Union[str, os.PathLike]) -> Path:
        local_path = Path(local_path)
        local_path.parent.mkdir(exist_ok=True, parents=True)

        try:
            shutil.copyfile(self._cloud_path_to_local(cloud_path), local_path)
        except FileNotFoundError:
            # erroneous FileNotFoundError appears in tests sometimes; patiently insist on the parent directory existing
            sleep(1.0)
            local_path.parent.mkdir(exist_ok=True, parents=True)
            sleep(1.0)

            shutil.copyfile(self._cloud_path_to_local(cloud_path), local_path)

        return local_path

    def _exists(self, cloud_path: "LocalPath") -> bool:
        return self._cloud_path_to_local(cloud_path).exists()

    def _is_dir(self, cloud_path: "LocalPath") -> bool:
        return self._cloud_path_to_local(cloud_path).is_dir()

    def _is_file(self, cloud_path: "LocalPath") -> bool:
        return self._cloud_path_to_local(cloud_path).is_file()

    def _list_dir(
        self, cloud_path: "LocalPath", recursive=False
    ) -> Iterable[Tuple["LocalPath", bool]]:
        pattern = "**/*" if recursive else "*"
        for obj in self._cloud_path_to_local(cloud_path).glob(pattern):
            yield (self._local_to_cloud_path(obj), obj.is_dir())

    def _md5(self, cloud_path: "LocalPath") -> str:
        return md5(self._cloud_path_to_local(cloud_path).read_bytes()).hexdigest()

    def _move_file(
        self, src: "LocalPath", dst: "LocalPath", remove_src: bool = True
    ) -> "LocalPath":
        self._cloud_path_to_local(dst).parent.mkdir(exist_ok=True, parents=True)

        if remove_src:
            self._cloud_path_to_local(src).replace(self._cloud_path_to_local(dst))
        else:
            shutil.copy(self._cloud_path_to_local(src), self._cloud_path_to_local(dst))
        return dst

    def _remove(self, cloud_path: "LocalPath", missing_ok: bool = True) -> None:
        local_storage_path = self._cloud_path_to_local(cloud_path)
        if not missing_ok and not local_storage_path.exists():
            raise FileNotFoundError(f"File does not exist: {cloud_path}")

        if local_storage_path.is_file():
            local_storage_path.unlink()
        elif local_storage_path.is_dir():
            shutil.rmtree(local_storage_path)

    def _stat(self, cloud_path: "LocalPath") -> os.stat_result:
        stat_result = self._cloud_path_to_local(cloud_path).stat()

        return os.stat_result(
            (  # type: ignore
                None,  # type: ignore # mode
                None,  # ino
                cloud_path.cloud_prefix,  # dev,
                None,  # nlink,
                None,  # uid,
                None,  # gid,
                stat_result.st_size,  # size,
                None,  # atime,
                stat_result.st_mtime,  # mtime,
                None,  # ctime,
            )
        )

    def _touch(self, cloud_path: "LocalPath", exist_ok: bool = True) -> None:
        local_storage_path = self._cloud_path_to_local(cloud_path)
        if local_storage_path.exists() and not exist_ok:
            raise FileExistsError(f"File exists: {cloud_path}")
        local_storage_path.parent.mkdir(exist_ok=True, parents=True)
        local_storage_path.touch()

    def _upload_file(
        self, local_path: Union[str, os.PathLike], cloud_path: "LocalPath"
    ) -> "LocalPath":
        dst = self._cloud_path_to_local(cloud_path)
        dst.parent.mkdir(exist_ok=True, parents=True)
        shutil.copy(local_path, dst)
        return cloud_path

    def _get_metadata(self, cloud_path: "LocalPath") -> Dict:
        # content_type is the only metadata we test currently
        if self.content_type_method is None:
            content_type_method = lambda x: (None, None)
        else:
            content_type_method = self.content_type_method

        return {
            "content_type": content_type_method(str(self._cloud_path_to_local(cloud_path)))[0],
        }

    def _get_public_url(self, cloud_path: "LocalPath") -> str:
        return cloud_path.as_uri()

    def _generate_presigned_url(
        self, cloud_path: "LocalPath", expire_seconds: int = 60 * 60
    ) -> str:
        raise NotImplementedError("Cannot generate a presigned URL for a local path.")

Attributes

local_storage_dir: Path property readonly

The local directory where files are stored for this client. This storage directory is the one that simulates the cloud. If no storage directory was provided on instantiating the client, the default storage directory for this client class is used.

Methods

__init__(self, *args, *, local_storage_dir: Union[str, os.PathLike] = None, file_cache_mode: Union[str, cloudpathlib.enums.FileCacheMode] = None, local_cache_dir: Union[str, os.PathLike] = None, content_type_method: Optional[Callable] = <function guess_type at 0x7f9a1cd285e0>, **kwargs) special

Source code in cloudpathlib/local/localclient.py

def __init__(
    self,
    *args,
    local_storage_dir: Optional[Union[str, os.PathLike]] = None,
    file_cache_mode: Optional[Union[str, FileCacheMode]] = None,
    local_cache_dir: Optional[Union[str, os.PathLike]] = None,
    content_type_method: Optional[Callable] = mimetypes.guess_type,
    **kwargs,
):
    self._local_storage_dir = local_storage_dir

    super().__init__(
        local_cache_dir=local_cache_dir,
        content_type_method=content_type_method,
        file_cache_mode=file_cache_mode,
    )

get_default_storage_dir() -> Path classmethod

Return the default storage directory for this client class. This is used if a client is instantiated without a storage directory being explicitly provided. In this usage, "storage" refers to the local storage that simulates the cloud.

Source code in cloudpathlib/local/localclient.py

@classmethod
def get_default_storage_dir(cls) -> Path:
    """Return the default storage directory for this client class. This is used if a client
    is instantiated without a storage directory being explicitly provided. In this usage,
    "storage" refers to the local storage that simulates the cloud.
    """
    if cls._default_storage_temp_dir is None:
        cls._default_storage_temp_dir = TemporaryDirectory()
        _temp_dirs_to_clean.append(cls._default_storage_temp_dir)
    return Path(cls._default_storage_temp_dir.name)

reset_default_storage_dir() -> Path classmethod

Reset the default storage directly. This tears down and recreates the directory used by default for this client class when instantiating a client without explicitly providing a storage directory. In this usage, "storage" refers to the local storage that simulates the cloud.

Source code in cloudpathlib/local/localclient.py

@classmethod
def reset_default_storage_dir(cls) -> Path:
    """Reset the default storage directly. This tears down and recreates the directory used by
    default for this client class when instantiating a client without explicitly providing
    a storage directory. In this usage, "storage" refers to the local storage that simulates
    the cloud.
    """
    cls._default_storage_temp_dir = None
    return cls.get_default_storage_dir()

Replacement for GSClient that uses the local file system. Intended as a monkeypatch substitute when writing tests.

Source code in cloudpathlib/local/implementations/gs.py

class LocalGSClient(LocalClient):
    """Replacement for GSClient that uses the local file system. Intended as a monkeypatch
    substitute when writing tests.
    """

    _cloud_meta = local_gs_implementation

GSPath(self, cloud_path: Union[str, ~BoundedCloudPath]) -> ~BoundedCloudPath

Source code in cloudpathlib/local/implementations/gs.py

def CloudPath(self, cloud_path: Union[str, BoundedCloudPath]) -> BoundedCloudPath:
    return self._cloud_meta.path_class(cloud_path=cloud_path, client=self)  # type: ignore

Replacement for GSPath that uses the local file system. Intended as a monkeypatch substitute when writing tests.

Attributes

blob: str property readonly

bucket: str property readonly

cloud_prefix: str

drive: str property readonly

The drive prefix (letter or UNC path), if any. (Docstring copied from pathlib.Path)

etag property readonly

Methods

mkdir(self, parents = False, exist_ok = False)

Create a new directory at this given path. (Docstring copied from pathlib.Path)

Source code in cloudpathlib/local/implementations/gs.py

def mkdir(self, parents=False, exist_ok=False):
    # not possible to make empty directory on gs
    pass

Abstract CloudPath for accessing objects the local filesystem. Subclasses are as a monkeypatch substitutes for normal CloudPath subclasses when writing tests.

Source code in cloudpathlib/local/localpath.py

class LocalPath(CloudPath):
    """Abstract CloudPath for accessing objects the local filesystem. Subclasses are as a
    monkeypatch substitutes for normal CloudPath subclasses when writing tests."""

    client: "LocalClient"

    def is_dir(self) -> bool:
        return self.client._is_dir(self)

    def is_file(self) -> bool:
        return self.client._is_file(self)

    def stat(self):
        try:
            meta = self.client._stat(self)
        except FileNotFoundError:
            raise NoStatError(
                f"No stats available for {self}; it may be a directory or not exist."
            )
        return meta

    def touch(self, exist_ok: bool = True):
        self.client._touch(self, exist_ok)

Methods

is_dir(self) -> bool

Whether this path is a directory. (Docstring copied from pathlib.Path)

Source code in cloudpathlib/local/localpath.py

def is_dir(self) -> bool:
    return self.client._is_dir(self)

is_file(self) -> bool

Whether this path is a regular file (also True for symlinks pointing to regular files). (Docstring copied from pathlib.Path)

Source code in cloudpathlib/local/localpath.py

def is_file(self) -> bool:
    return self.client._is_file(self)

stat(self)

Return the result of the stat() system call on this path, like os.stat() does. (Docstring copied from pathlib.Path)

Source code in cloudpathlib/local/localpath.py

def stat(self):
    try:
        meta = self.client._stat(self)
    except FileNotFoundError:
        raise NoStatError(
            f"No stats available for {self}; it may be a directory or not exist."
        )
    return meta

touch(self, exist_ok: bool = True)

Create this file with the given access mode, if it doesn't exist. (Docstring copied from pathlib.Path)

Source code in cloudpathlib/local/localpath.py

def touch(self, exist_ok: bool = True):
    self.client._touch(self, exist_ok)

Replacement for S3Client that uses the local file system. Intended as a monkeypatch substitute when writing tests.

Source code in cloudpathlib/local/implementations/s3.py

class LocalS3Client(LocalClient):
    """Replacement for S3Client that uses the local file system. Intended as a monkeypatch
    substitute when writing tests.
    """

    _cloud_meta = local_s3_implementation

S3Path(self, cloud_path: Union[str, ~BoundedCloudPath]) -> ~BoundedCloudPath

Source code in cloudpathlib/local/implementations/s3.py

def CloudPath(self, cloud_path: Union[str, BoundedCloudPath]) -> BoundedCloudPath:
    return self._cloud_meta.path_class(cloud_path=cloud_path, client=self)  # type: ignore

Replacement for S3Path that uses the local file system. Intended as a monkeypatch substitute when writing tests.

Attributes

bucket: str property readonly

cloud_prefix: str

drive: str property readonly

The drive prefix (letter or UNC path), if any. (Docstring copied from pathlib.Path)

etag property readonly

key: str property readonly

Methods

mkdir(self, parents = False, exist_ok = False)

Create a new directory at this given path. (Docstring copied from pathlib.Path)

Source code in cloudpathlib/local/implementations/s3.py

def mkdir(self, parents=False, exist_ok=False):
    # not possible to make empty directory on s3
    pass