Skip to content

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.

Attributes

local_azure_blob_implementation: None

local_gs_implementation: None

local_s3_implementation: None

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),
        ]
        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."
            )
        super().__init__(*args, **kwargs)
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)
__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),
    ]
    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."
        )
    super().__init__(*args, **kwargs)

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

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_cache_dir: Optional[Union[str, os.PathLike]] = None,
        local_storage_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)

    @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)
        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],
        }
__init__(self, *args, *, local_cache_dir: Union[str, os.PathLike] = None, local_storage_dir: Union[str, os.PathLike] = None, content_type_method: Optional[Callable] = <function guess_type at 0x7f6e8202e790>, **kwargs) special
Source code in cloudpathlib/local/localclient.py
def __init__(
    self,
    *args,
    local_cache_dir: Optional[Union[str, os.PathLike]] = None,
    local_storage_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)
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()

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)

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

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)

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)

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