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.

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.

Source code in cloudpathlib/local/implementations/azure.py 
class LocalAzureBlobPath(LocalPath):
    """Replacement for AzureBlobPath that uses the local file system. Intended as a monkeypatch
    substitute when writing tests.
    """

    cloud_prefix: str = "az://"
    _cloud_meta = local_azure_blob_implementation

    @property
    def drive(self) -> str:
        return self.container

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

    @property
    def container(self) -> str:
        return self._no_prefix.split("/", 1)[0]

    @property
    def blob(self) -> str:
        key = self._no_prefix_no_drive

        # key should never have starting slash for
        if key.startswith("/"):
            key = key[1:]

        return key

    @property
    def etag(self):
        return self.client._md5(self)

    @property
    def md5(self) -> str:
        return self.client._md5(self)
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.

Source code in cloudpathlib/local/implementations/gs.py 
class LocalGSPath(LocalPath):
    """Replacement for GSPath that uses the local file system. Intended as a monkeypatch substitute
    when writing tests.
    """

    cloud_prefix: str = "gs://"
    _cloud_meta = local_gs_implementation

    @property
    def drive(self) -> str:
        return self.bucket

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

    @property
    def bucket(self) -> str:
        return self._no_prefix.split("/", 1)[0]

    @property
    def blob(self) -> str:
        key = self._no_prefix_no_drive

        # key should never have starting slash for
        # use with boto, etc.
        if key.startswith("/"):
            key = key[1:]

        return key

    @property
    def etag(self):
        return self.client._md5(self)
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.

Source code in cloudpathlib/local/implementations/s3.py 
class LocalS3Path(LocalPath):
    """Replacement for S3Path that uses the local file system. Intended as a monkeypatch substitute
    when writing tests.
    """

    cloud_prefix: str = "s3://"
    _cloud_meta = local_s3_implementation

    @property
    def drive(self) -> str:
        return self.bucket

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

    @property
    def bucket(self) -> str:
        return self._no_prefix.split("/", 1)[0]

    @property
    def key(self) -> str:
        key = self._no_prefix_no_drive

        # key should never have starting slash for
        # use with boto, etc.
        if key.startswith("/"):
            key = key[1:]

        return key

    @property
    def etag(self):
        return self.client._md5(self)
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", follow_symlinks=True) -> bool:
        kwargs = dict(follow_symlinks=follow_symlinks)
        if sys.version_info < (3, 13):
            kwargs.pop("follow_symlinks")

        return self._cloud_path_to_local(cloud_path).is_dir(**kwargs)

    def _is_file(self, cloud_path: "LocalPath", follow_symlinks=True) -> bool:
        kwargs = dict(follow_symlinks=follow_symlinks)
        if sys.version_info < (3, 13):
            kwargs.pop("follow_symlinks")

        return self._cloud_path_to_local(cloud_path).is_file(**kwargs)

    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 0x7feff44589a0>, **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, follow_symlinks=True) -> bool:
        return self.client._is_dir(self, follow_symlinks=follow_symlinks)

    def is_file(self, follow_symlinks=True) -> bool:
        return self.client._is_file(self, follow_symlinks=follow_symlinks)

    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, follow_symlinks = True) -> bool

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

Source code in cloudpathlib/local/localpath.py 
def is_dir(self, follow_symlinks=True) -> bool:
    return self.client._is_dir(self, follow_symlinks=follow_symlinks)
is_file(self, follow_symlinks = True) -> 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, follow_symlinks=True) -> bool:
    return self.client._is_file(self, follow_symlinks=follow_symlinks)
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.

Source code in cloudpathlib/local/implementations/azure.py 
class LocalAzureBlobPath(LocalPath):
    """Replacement for AzureBlobPath that uses the local file system. Intended as a monkeypatch
    substitute when writing tests.
    """

    cloud_prefix: str = "az://"
    _cloud_meta = local_azure_blob_implementation

    @property
    def drive(self) -> str:
        return self.container

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

    @property
    def container(self) -> str:
        return self._no_prefix.split("/", 1)[0]

    @property
    def blob(self) -> str:
        key = self._no_prefix_no_drive

        # key should never have starting slash for
        if key.startswith("/"):
            key = key[1:]

        return key

    @property
    def etag(self):
        return self.client._md5(self)

    @property
    def md5(self) -> str:
        return self.client._md5(self)

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", follow_symlinks=True) -> bool:
        kwargs = dict(follow_symlinks=follow_symlinks)
        if sys.version_info < (3, 13):
            kwargs.pop("follow_symlinks")

        return self._cloud_path_to_local(cloud_path).is_dir(**kwargs)

    def _is_file(self, cloud_path: "LocalPath", follow_symlinks=True) -> bool:
        kwargs = dict(follow_symlinks=follow_symlinks)
        if sys.version_info < (3, 13):
            kwargs.pop("follow_symlinks")

        return self._cloud_path_to_local(cloud_path).is_file(**kwargs)

    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 0x7feff44589a0>, **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.

Source code in cloudpathlib/local/implementations/gs.py 
class LocalGSPath(LocalPath):
    """Replacement for GSPath that uses the local file system. Intended as a monkeypatch substitute
    when writing tests.
    """

    cloud_prefix: str = "gs://"
    _cloud_meta = local_gs_implementation

    @property
    def drive(self) -> str:
        return self.bucket

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

    @property
    def bucket(self) -> str:
        return self._no_prefix.split("/", 1)[0]

    @property
    def blob(self) -> str:
        key = self._no_prefix_no_drive

        # key should never have starting slash for
        # use with boto, etc.
        if key.startswith("/"):
            key = key[1:]

        return key

    @property
    def etag(self):
        return self.client._md5(self)

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, follow_symlinks=True) -> bool:
        return self.client._is_dir(self, follow_symlinks=follow_symlinks)

    def is_file(self, follow_symlinks=True) -> bool:
        return self.client._is_file(self, follow_symlinks=follow_symlinks)

    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, follow_symlinks = True) -> bool

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

Source code in cloudpathlib/local/localpath.py 
def is_dir(self, follow_symlinks=True) -> bool:
    return self.client._is_dir(self, follow_symlinks=follow_symlinks)

is_file(self, follow_symlinks = True) -> 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, follow_symlinks=True) -> bool:
    return self.client._is_file(self, follow_symlinks=follow_symlinks)

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.

Source code in cloudpathlib/local/implementations/s3.py 
class LocalS3Path(LocalPath):
    """Replacement for S3Path that uses the local file system. Intended as a monkeypatch substitute
    when writing tests.
    """

    cloud_prefix: str = "s3://"
    _cloud_meta = local_s3_implementation

    @property
    def drive(self) -> str:
        return self.bucket

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

    @property
    def bucket(self) -> str:
        return self._no_prefix.split("/", 1)[0]

    @property
    def key(self) -> str:
        key = self._no_prefix_no_drive

        # key should never have starting slash for
        # use with boto, etc.
        if key.startswith("/"):
            key = key[1:]

        return key

    @property
    def etag(self):
        return self.client._md5(self)

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