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

@register_path_class("azure")
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 mkdir(self, parents=False, exist_ok=False):
        self.client._mkdir(self, parents=parents, exist_ok=exist_ok)

    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,
            )
        )

    def replace(self, target: "AzureBlobPath") -> "AzureBlobPath":
        try:
            return super().replace(target)

        # we can rename directories on ADLS Gen2
        except CloudPathIsADirectoryError:
            if self.client._check_hns(self):
                return self.client._move_file(self, target)
            else:
                raise

    @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

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):
    self.client._mkdir(self, parents=parents, exist_ok=exist_ok)

replace(self, target: AzureBlobPath) -> AzureBlobPath

Rename this path to the target path, overwriting if that path exists.

The target path may be absolute or relative. Relative paths are interpreted relative to the current working directory, not the directory of the Path object.

Returns the new Path instance pointing to the target path. (Docstring copied from pathlib.Path)

Source code in cloudpathlib/azure/azblobpath.py

def replace(self, target: "AzureBlobPath") -> "AzureBlobPath":
    try:
        return super().replace(target)

    # we can rename directories on ADLS Gen2
    except CloudPathIsADirectoryError:
        if self.client._check_hns(self):
            return self.client._move_file(self, target)
        else:
            raise

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()