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."""

    _default_storage_temp_dir = None

    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,
    ):
        # setup caching and local versions of file. use default temp dir if not provided
        if local_storage_dir is None:
            local_storage_dir = self.get_default_storage_dir()
        self._local_storage_dir = Path(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:
        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:
        cls._default_storage_temp_dir = None
        return cls.get_default_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]]:
        if recursive:
            return (
                (self._local_to_cloud_path(obj), obj.is_dir())
                for obj in self._cloud_path_to_local(cloud_path).glob("**/*")
            )
        return (
            (self._local_to_cloud_path(obj), obj.is_dir())
            for obj in self._cloud_path_to_local(cloud_path).iterdir()
        )

    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.")

__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 0x7fd2bd1534c0>, **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,
):
    # setup caching and local versions of file. use default temp dir if not provided
    if local_storage_dir is None:
        local_storage_dir = self.get_default_storage_dir()
    self._local_storage_dir = Path(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

Source code in cloudpathlib/local/localclient.py

@classmethod
def get_default_storage_dir(cls) -> Path:
    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

Source code in cloudpathlib/local/localclient.py

@classmethod
def reset_default_storage_dir(cls) -> Path:
    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."""

    _default_storage_temp_dir = None

    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,
    ):
        # setup caching and local versions of file. use default temp dir if not provided
        if local_storage_dir is None:
            local_storage_dir = self.get_default_storage_dir()
        self._local_storage_dir = Path(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:
        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:
        cls._default_storage_temp_dir = None
        return cls.get_default_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]]:
        if recursive:
            return (
                (self._local_to_cloud_path(obj), obj.is_dir())
                for obj in self._cloud_path_to_local(cloud_path).glob("**/*")
            )
        return (
            (self._local_to_cloud_path(obj), obj.is_dir())
            for obj in self._cloud_path_to_local(cloud_path).iterdir()
        )

    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.")

__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 0x7fd2bd1534c0>, **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,
):
    # setup caching and local versions of file. use default temp dir if not provided
    if local_storage_dir is None:
        local_storage_dir = self.get_default_storage_dir()
    self._local_storage_dir = Path(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

Source code in cloudpathlib/local/localclient.py

@classmethod
def get_default_storage_dir(cls) -> Path:
    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

Source code in cloudpathlib/local/localclient.py

@classmethod
def reset_default_storage_dir(cls) -> Path:
    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