cloudpathlib.AzureBlobPath

Class for representing and operating on Azure Blob Storage URIs, in the style of the Python standard library's pathlib module. Instances represent a path in Blob Storage with filesystem path semantics, and convenient methods allow for basic operations like joining, reading, writing, iterating over contents, etc. This class almost entirely mimics the pathlib.Path interface, so most familiar properties and methods should be available and behave in the expected way.

The AzureBlobClient class handles authentication with Azure. If a client instance is not explicitly specified on AzureBlobPath instantiation, a default client is used. See AzureBlobClient's documentation for more details.

Source code in cloudpathlib/azure/azblobpath.py

class AzureBlobPath(CloudPath):
    """Class for representing and operating on Azure Blob Storage URIs, in the style of the Python
    standard library's [`pathlib` module](https://docs.python.org/3/library/pathlib.html).
    Instances represent a path in Blob Storage with filesystem path semantics, and convenient
    methods allow for basic operations like joining, reading, writing, iterating over contents,
    etc. This class almost entirely mimics the [`pathlib.Path`](https://docs.python.org/3/library/pathlib.html#pathlib.Path)
    interface, so most familiar properties and methods should be available and behave in the
    expected way.

    The [`AzureBlobClient`](../azblobclient/) class handles authentication with Azure. If a
    client instance is not explicitly specified on `AzureBlobPath` instantiation, a default client
    is used. See `AzureBlobClient`'s documentation for more details.
    """

    cloud_prefix: str = "az://"
    client: "AzureBlobClient"

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

    def is_dir(self) -> bool:
        return self.client._is_file_or_dir(self) == "dir"

    def is_file(self) -> bool:
        return self.client._is_file_or_dir(self) == "file"

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

    def touch(self, exist_ok: bool = True):
        if self.exists():
            if not exist_ok:
                raise FileExistsError(f"File exists: {self}")
            self.client._move_file(self, self)
        else:
            tf = TemporaryDirectory()
            p = Path(tf.name) / "empty"
            p.touch()

            self.client._upload_file(p, self)

            tf.cleanup()

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

        return os.stat_result(
            (
                None,  # mode
                None,  # ino
                self.cloud_prefix,  # dev,
                None,  # nlink,
                None,  # uid,
                None,  # gid,
                meta.get("size", 0),  # size,
                None,  # atime,
                meta.get("last_modified", 0).timestamp(),  # mtime,
                None,  # ctime,
            )
        )

    @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._get_metadata(self).get("etag", None)

    @property
    def md5(self) -> str:
        return self.client._get_metadata(self).get("content_settings", {}).get("content_md5", None)

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

is_dir(self) -> bool

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

Source code in cloudpathlib/azure/azblobpath.py

def is_dir(self) -> bool:
    return self.client._is_file_or_dir(self) == "dir"

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/azure/azblobpath.py

def is_file(self) -> bool:
    return self.client._is_file_or_dir(self) == "file"

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

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

Source code in cloudpathlib/azure/azblobpath.py

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

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/azure/azblobpath.py

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

    return os.stat_result(
        (
            None,  # mode
            None,  # ino
            self.cloud_prefix,  # dev,
            None,  # nlink,
            None,  # uid,
            None,  # gid,
            meta.get("size", 0),  # size,
            None,  # atime,
            meta.get("last_modified", 0).timestamp(),  # mtime,
            None,  # ctime,
        )
    )

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/azure/azblobpath.py

def touch(self, exist_ok: bool = True):
    if self.exists():
        if not exist_ok:
            raise FileExistsError(f"File exists: {self}")
        self.client._move_file(self, self)
    else:
        tf = TemporaryDirectory()
        p = Path(tf.name) / "empty"
        p.touch()

        self.client._upload_file(p, self)

        tf.cleanup()