s3:, gs: and az:

Each needs its own extra: s3, gs or az.

pathlib_next.uri.schemes.s3

BaseS3Backend

Bases: object

Protocol for obtaining a boto3 S3 client. Subclass this to plug in custom credential/session handling (e.g. tests mock it directly, no real AWS account needed); S3Backend is the real implementation.

client()

Source code in src/pathlib_next/uri/schemes/s3.py

@_utils.notimplemented
def client(self): ...

S3Backend(**client_kwargs)

Bases: BaseS3Backend

Lazily creates+caches a boto3 S3 client. Unlike sftp.py's/ ftp.py's per-thread connection pools, a single boto3 client is reused across threads -- it's documented as thread-safe.

Source code in src/pathlib_next/uri/schemes/s3.py

def __init__(self, **client_kwargs):
    self.client_kwargs = client_kwargs
    self._client = None

client_kwargs = client_kwargs instance-attribute

client()

Source code in src/pathlib_next/uri/schemes/s3.py

def client(self):
    if self._client is None:
        import boto3

        self._client = boto3.client("s3", **self.client_kwargs)
    return self._client

S3Path(*uris, **options)

Bases: UriPath

s3: scheme (s3://bucket/key/path): read/write/list via boto3. Requires the s3 extra. S3 has no real directories -- is_dir() is prefix emulation (any object key under "<path>/"), and mkdir() creates a zero-byte "<path>/" marker object (the same convention the AWS console itself uses for an empty "folder"); see docs/divergences.md. rename() uses server-side copy_object + delete_object (same-bucket only) instead of the generic download+upload+delete move() fallback; a prefix directory is not renamed (NotImplementedError), so move() copies and removes it.

Source code in src/pathlib_next/uri/__init__.py

def __init__(self, *uris: UriLike, **options):
    if self._raw_uris or self._initiated:
        return
    _uris: list[str | Uri] = []
    for uri in uris:
        if not uri:
            uri = ""
        if isinstance(uri, Uri):
            _uris.append(uri)
        elif isinstance(uri, (_pathlib.Path, Path)):
            try:
                uri = uri.as_uri()
            except ValueError:
                # as_uri() raises ValueError for a relative path, which
                # joins like a relative PurePath (see _RelativeLocalPath).
                uri = _RelativeLocalPath(_uriencode(uri.as_posix(), safe="/"))
            _uris.append(uri)
        elif isinstance(uri, (_pathlib.PurePath, Pathname)):
            _uris.append(_path_reference(uri.as_posix()))
        elif hasattr(uri, "as_uri"):
            path = uri.as_uri
            if callable(path):
                path = path()
            _uris.append(path)
        elif isinstance(uri, str):
            _uris.append(uri)
        elif isinstance(uri, bytes):
            _uris.append(uri.decode())
        else:
            path = None
            try:
                path = os.fspath(uri)
            except (TypeError, NotImplementedError):
                pass
            if not isinstance(path, str):
                raise TypeError(
                    "argument should be a str or an os.PathLike "
                    "object where __fspath__ returns a str, "
                    f"not {type(path).__name__!r}"
                )
            # Only __fspath__ is guaranteed here -- posix-normalize the
            # string itself rather than assuming an as_posix() method.
            posix = _pathlib.PurePath(path).as_posix()
            _uris.append(_path_reference(posix))
    self._raw_uris = _uris

backend instance-attribute

bucket property

key property

rename(target)

Source code in src/pathlib_next/uri/schemes/s3.py

def rename(self, target: "S3Path | Uri | str"):
    target = self._rename_target(target)
    dest_key = _object_key(target.path)
    # pathlib returns the new path.
    renamed = self.with_path(target.path)
    if dest_key == self.key:
        return renamed
    client = self._client
    try:
        head = client.head_object(Bucket=self.bucket, Key=self.key)
    except _S3_ERRORS as error:
        if not (isinstance(error, _botoexc.ClientError) and _is_not_found(error)):
            raise _oserror(error, self) from error
        head = None
    if head is None:
        # No object at the key: a prefix directory (stat() raises
        # FileNotFoundError when there is nothing at all). Its keys are
        # not renamed one by one here; move() falls back to copy + rm.
        self._pop_stat_hint()
        self.stat()
        raise NotImplementedError(f"rename() of the prefix directory {self}")
    extra = {}
    storage_class = head.get("StorageClass")
    if storage_class and storage_class != "STANDARD":
        # CopyObject writes STANDARD unless told otherwise.
        extra["StorageClass"] = storage_class
    source = {"Bucket": self.bucket, "Key": self.key}
    managed_copy = getattr(client, "copy", None)
    try:
        if head.get("ContentLength", 0) > _PUT_OBJECT_LIMIT and callable(
            managed_copy
        ):
            # CopyObject refuses sources over 5 GB; the managed copy
            # uses UploadPartCopy.
            managed_copy(source, self.bucket, dest_key, ExtraArgs=extra or None)
        else:
            client.copy_object(
                Bucket=self.bucket, Key=dest_key, CopySource=source, **extra
            )
        client.delete_object(Bucket=self.bucket, Key=self.key)
    except _S3_ERRORS as error:
        raise _oserror(error, self) from error
    return renamed

rm(recursive=False, missing_ok=False, ignore_error=False)

Source code in src/pathlib_next/uri/schemes/s3.py

def rm(
    self,
    /,
    recursive=False,
    missing_ok=False,
    ignore_error: bool | _ty.Callable[[Exception, _ty.Self], bool] = False,
):
    if not recursive:
        return super().rm(
            recursive=recursive,
            missing_ok=missing_ok,
            ignore_error=ignore_error,
        )

    def on_error(error):
        if callable(ignore_error):
            return ignore_error(error, self)
        return bool(ignore_error)

    if not self.key:
        error = PermissionError("recursive bucket delete is not enabled")
        if not on_error(error):
            raise error
        return

    keys = []
    try:
        self._client.head_object(Bucket=self.bucket, Key=self.key)
        keys.append(self.key)
    except _S3_ERRORS as error:
        if not (isinstance(error, _botoexc.ClientError) and _is_not_found(error)):
            translated = _oserror(error, self)
            if not on_error(translated):
                raise translated from error
            return
    if not keys:
        marker = f"{self.key}/" if self.key else ""
        paginator = self._client.get_paginator("list_objects_v2")
        try:
            for page in paginator.paginate(Bucket=self.bucket, Prefix=marker):
                keys.extend(obj["Key"] for obj in page.get("Contents", []))
        except _S3_ERRORS as error:
            translated = _oserror(error, self)
            if not on_error(translated):
                raise translated from error
            return

    if not keys:
        if missing_ok:
            return
        error = FileNotFoundError(self)
        if not on_error(error):
            raise error
        return

    iterator = iter(keys)
    while True:
        batch = list(_itertools.islice(iterator, 1000))
        if not batch:
            break
        try:
            try:
                response = self._client.delete_objects(
                    Bucket=self.bucket,
                    Delete={"Objects": [{"Key": key} for key in batch]},
                )
            except _S3_ERRORS as error:
                raise _oserror(error, self) from error
            errors = response.get("Errors", []) if response else []
            if errors:
                raise OSError(f"S3 delete_objects failed: {errors!r}")
        except Exception as error:
            if not on_error(error):
                raise

rmdir()

Source code in src/pathlib_next/uri/schemes/s3.py

def rmdir(self):
    if not self.stat().is_dir():
        raise self._not_a_directory()
    if not self.key:
        raise PermissionError(
            _errno.EACCES, "removing a bucket is not supported", str(self)
        )
    marker = f"{self.key}/"
    try:
        resp = self._client.list_objects_v2(
            Bucket=self.bucket, Prefix=marker, MaxKeys=2
        )
        contents = resp.get("Contents", [])
        if any(obj["Key"] != marker for obj in contents):
            raise OSError(_errno.ENOTEMPTY, "Directory not empty", str(self))
        self._client.delete_object(Bucket=self.bucket, Key=marker)
    except _S3_ERRORS as error:
        raise _oserror(error, self) from error

stat(*, follow_symlinks=True)

Source code in src/pathlib_next/uri/schemes/s3.py

def stat(self, *, follow_symlinks=True):
    hint = self._pop_stat_hint()
    if hint is not None:
        return hint
    key = self.key
    client = self._client
    if key == "":
        try:
            client.head_bucket(Bucket=self.bucket)
        except _S3_ERRORS as error:
            raise _oserror(error, self) from error
        return FileStat(is_dir=True)
    try:
        head = client.head_object(Bucket=self.bucket, Key=key)
    except _botoexc.ClientError as error:
        # A 403 stays PermissionError: S3 answers HEAD of a missing key
        # with 403 when the caller may not list the bucket.
        if not _is_not_found(error):
            raise _oserror(error, self) from error
    except _TRANSPORT_ERRORS as error:
        raise _oserror(error, self) from error
    else:
        return FileStat(
            st_size=head["ContentLength"],
            st_mtime=int(head["LastModified"].timestamp()),
            is_dir=False,
        )
    # Not an object at this exact key -- emulate a directory: any
    # object under the "<key>/" prefix means this is a "directory".
    try:
        resp = client.list_objects_v2(
            Bucket=self.bucket, Prefix=f"{key}/", MaxKeys=1
        )
    except _S3_ERRORS as error:
        raise _oserror(error, self) from error
    if resp.get("KeyCount", 0) > 0:
        return FileStat(is_dir=True)
    raise FileNotFoundError(self)

unlink(missing_ok=False)

Source code in src/pathlib_next/uri/schemes/s3.py

def unlink(self, missing_ok=False):
    try:
        st = self.stat()
    except FileNotFoundError:
        if missing_ok:
            return
        raise
    if st.is_dir():
        # A prefix: delete_object() of the bare key would delete nothing
        # and report success.
        raise IsADirectoryError(_errno.EISDIR, "Is a directory", str(self))
    try:
        self._client.delete_object(Bucket=self.bucket, Key=self.key)
    except _S3_ERRORS as error:
        raise _oserror(error, self) from error

pathlib_next.uri.schemes.gs

BaseGsBackend

Bases: object

Protocol for obtaining a google.cloud.storage Client. Subclass this to plug in custom credential/session handling (e.g. tests can override to point at a fake server); GsBackend is the real implementation.

client()

Source code in src/pathlib_next/uri/schemes/gs.py

@_utils.notimplemented
def client(self): ...

GsBackend(**client_kwargs)

Bases: BaseGsBackend

Lazily creates+caches a google.cloud.storage Client. A single client is reused across threads -- it's documented as thread-safe.

client_kwargs go to storage.Client unchanged, client_options (dict or ClientOptions, api_endpoint included) among them. A custom endpoint still authenticates; for an emulator or fake server pass use_auth_w_custom_endpoint=False (anonymous credentials), or set STORAGE_EMULATOR_HOST yourself -- the SDK reads it.

Source code in src/pathlib_next/uri/schemes/gs.py

def __init__(self, **client_kwargs):
    self.client_kwargs = client_kwargs
    self._client = None

client_kwargs = client_kwargs instance-attribute

client()

Source code in src/pathlib_next/uri/schemes/gs.py

def client(self):
    if self._client is None:
        from google.cloud import storage

        # Passed through as given. This used to turn a dict
        # `api_endpoint` into a process-wide, never-restored
        # `os.environ["STORAGE_EMULATOR_HOST"]` and drop the other
        # client options: every later client in the process (and any
        # subprocess) was redirected to that endpoint with anonymous
        # credentials.
        self._client = storage.Client(**self.client_kwargs)
    return self._client

GsPath(*uris, **options)

Bases: UriPath

gs: scheme (gs://bucket/key/path): read/write/list via google.cloud.storage. Requires the gs extra. GCS has no real directories -- is_dir() is prefix emulation (any object key under "<path>/"), and mkdir() creates a zero-byte "<path>/" marker object; see docs/divergences.md. rename() uses server-side copy + delete (same-bucket only) instead of the generic download+upload+delete move() fallback; a prefix directory is not renamed (NotImplementedError), so move() copies and removes it.

Source code in src/pathlib_next/uri/__init__.py

def __init__(self, *uris: UriLike, **options):
    if self._raw_uris or self._initiated:
        return
    _uris: list[str | Uri] = []
    for uri in uris:
        if not uri:
            uri = ""
        if isinstance(uri, Uri):
            _uris.append(uri)
        elif isinstance(uri, (_pathlib.Path, Path)):
            try:
                uri = uri.as_uri()
            except ValueError:
                # as_uri() raises ValueError for a relative path, which
                # joins like a relative PurePath (see _RelativeLocalPath).
                uri = _RelativeLocalPath(_uriencode(uri.as_posix(), safe="/"))
            _uris.append(uri)
        elif isinstance(uri, (_pathlib.PurePath, Pathname)):
            _uris.append(_path_reference(uri.as_posix()))
        elif hasattr(uri, "as_uri"):
            path = uri.as_uri
            if callable(path):
                path = path()
            _uris.append(path)
        elif isinstance(uri, str):
            _uris.append(uri)
        elif isinstance(uri, bytes):
            _uris.append(uri.decode())
        else:
            path = None
            try:
                path = os.fspath(uri)
            except (TypeError, NotImplementedError):
                pass
            if not isinstance(path, str):
                raise TypeError(
                    "argument should be a str or an os.PathLike "
                    "object where __fspath__ returns a str, "
                    f"not {type(path).__name__!r}"
                )
            # Only __fspath__ is guaranteed here -- posix-normalize the
            # string itself rather than assuming an as_posix() method.
            posix = _pathlib.PurePath(path).as_posix()
            _uris.append(_path_reference(posix))
    self._raw_uris = _uris

backend instance-attribute

bucket_name property

key property

rename(target)

Source code in src/pathlib_next/uri/schemes/gs.py

def rename(self, target: "GsPath | Uri | str"):
    target = self._rename_target(target)
    dest_key = _object_key(target.path)
    # pathlib returns the new path.
    renamed = self.with_path(target.path)
    if dest_key == self.key:
        # Copying onto itself and then deleting the source loses the object.
        return renamed
    source_blob = self._reload(self.key)
    if source_blob is None:
        # No object at the key: a prefix directory (stat() raises
        # FileNotFoundError when there is nothing at all). move() falls
        # back to copy + rm for it.
        self._pop_stat_hint()
        self.stat()
        raise NotImplementedError(f"rename() of the prefix directory {self}")
    with _translate_errors(self):
        self._bucket.copy_blob(source_blob, self._bucket, dest_key)
        source_blob.delete()
    return renamed

rm(recursive=False, missing_ok=False, ignore_error=False)

Source code in src/pathlib_next/uri/schemes/gs.py

def rm(
    self,
    /,
    recursive=False,
    missing_ok=False,
    ignore_error: bool | _ty.Callable[[Exception, _ty.Self], bool] = False,
):
    if not recursive:
        return super().rm(
            recursive=recursive,
            missing_ok=missing_ok,
            ignore_error=ignore_error,
        )

    def on_error(error):
        if callable(ignore_error):
            return ignore_error(error, self)
        return bool(ignore_error)

    if not self.key:
        error = PermissionError("recursive bucket delete is not enabled")
        if not on_error(error):
            raise error
        return

    keys = []
    try:
        if self._reload(self.key) is not None:
            keys.append(self.key)
    except OSError as error:
        # Not "missing": deleting the prefix tree instead of the object
        # would remove the wrong thing.
        if not on_error(error):
            raise
        return

    if not keys:
        marker = f"{self.key}/"
        try:
            with _translate_errors(self):
                keys.extend(
                    blob.name for blob in self._bucket.list_blobs(prefix=marker)
                )
        except Exception as error:
            if not on_error(error):
                raise
            return

    if not keys:
        if missing_ok:
            return
        error = FileNotFoundError(self)
        if not on_error(error):
            raise error
        return

    for key in keys:
        try:
            with _translate_errors(self):
                self._bucket.blob(key).delete()
        except Exception as error:
            if not on_error(error):
                raise

rmdir()

Source code in src/pathlib_next/uri/schemes/gs.py

def rmdir(self):
    if not self.stat().is_dir():
        raise NotADirectoryError(_errno.ENOTDIR, "Not a directory", str(self))
    marker = f"{self.key}/"
    with _translate_errors(self):
        for blob in self._bucket.list_blobs(prefix=marker, max_results=2):
            if blob.name != marker:
                raise OSError(_errno.ENOTEMPTY, "Directory not empty", str(self))
    try:
        with _translate_errors(self):
            self._bucket.blob(marker).delete()
    except FileNotFoundError:
        pass

stat(*, follow_symlinks=True)

Source code in src/pathlib_next/uri/schemes/gs.py

def stat(self, *, follow_symlinks=True):
    hint = self._pop_stat_hint()
    if hint is not None:
        return hint
    key = self.key
    if key == "":
        # The bucket root: ask the server, as s3: does with HeadBucket,
        # so a mistyped bucket does not read as an existing directory.
        # A one-item listing (NotFound for a missing bucket) needs only
        # the object-list permission the rest of the path uses.
        with _translate_errors(self):
            for _ in self._bucket.list_blobs(max_results=1):
                break
        return FileStat(is_dir=True)
    # Only a not-found reply falls through to the prefix probe: a
    # transient or permission error read as "missing" let copy() replace
    # an existing object.
    blob = self._reload(key)
    if blob is not None:
        return FileStat(
            st_size=blob.size,
            st_mtime=int(blob.updated.timestamp()) if blob.updated else 0,
            is_dir=False,
        )
    # Not an object at this exact key -- emulate a directory: any
    # object under the "<key>/" prefix means this is a "directory".
    prefix = f"{key}/"
    with _translate_errors(self):
        for _ in self._bucket.list_blobs(prefix=prefix, max_results=1):
            return FileStat(is_dir=True)
    raise FileNotFoundError(self)

unlink(missing_ok=False)

Source code in src/pathlib_next/uri/schemes/gs.py

def unlink(self, missing_ok=False):
    try:
        st = self.stat()
    except FileNotFoundError:
        if missing_ok:
            return
        raise
    if st.is_dir():
        raise IsADirectoryError(_errno.EISDIR, "Is a directory", str(self))
    blob = self._bucket.blob(self.key)
    try:
        with _translate_errors(self):
            blob.delete()
    except FileNotFoundError:
        # Deleted since the stat() above. Anything else -- a hold, a
        # permission error -- is not "already gone" and raises.
        if not missing_ok:
            raise

pathlib_next.uri.schemes.az

BaseAzBackend

Bases: object

Protocol for obtaining an azure.storage.blob.BlobServiceClient. Subclass this to plug in custom credential/session handling (e.g. tests can override to point at a fake server); AzBackend is the real implementation.

client()

Source code in src/pathlib_next/uri/schemes/az.py

@_utils.notimplemented
def client(self): ...

AzBackend(account=None, **client_kwargs)

Bases: BaseAzBackend

Lazily creates+caches an azure.storage.blob.BlobServiceClient. A single client is reused across threads -- it's documented as thread-safe.

client_kwargs go to BlobServiceClient unchanged; with connection_string, to BlobServiceClient.from_connection_string (credential and the other options included). account, when given and neither account_url nor connection_string is, derives account_url="https://<account>.blob.core.windows.net", authenticated by azure-identity's DefaultAzureCredential unless credential is passed -- the backend an AzPath built without backend= uses.

Source code in src/pathlib_next/uri/schemes/az.py

def __init__(self, account: "str | None" = None, **client_kwargs):
    self.client_kwargs = client_kwargs
    self.account = account
    self._client = None

account = account instance-attribute

client_kwargs = client_kwargs instance-attribute

client()

Source code in src/pathlib_next/uri/schemes/az.py

def client(self):
    if self._client is None:
        from azure.storage.blob import BlobServiceClient

        kwargs = dict(self.client_kwargs)
        if "connection_string" in kwargs:
            connection_string = kwargs.pop("connection_string")
            self._client = BlobServiceClient.from_connection_string(
                connection_string, **kwargs
            )
        else:
            if "account_url" not in kwargs and self.account:
                kwargs["account_url"] = (
                    f"https://{self.account}.blob.core.windows.net"
                )
                if "credential" not in kwargs:
                    kwargs["credential"] = _default_credential()
            self._client = BlobServiceClient(**kwargs)
    return self._client

AzPath(*uris, **options)

Bases: UriPath

az: scheme (az://account/container/key/path): read/write/list via azure.storage.blob. Requires the az extra. Azure Blob has no real directories -- is_dir() is prefix emulation (any blob key under "<path>/"), and mkdir() creates a zero-byte "<path>/" marker blob; see docs/divergences.md. rename() uses server-side copy + delete (same-container only) instead of the generic download+upload+delete move() fallback; a prefix directory is not renamed (NotImplementedError), so move() copies and removes it.

Without backend=, the client targets the URI's account (https://<account>.blob.core.windows.net) with azure-identity's DefaultAzureCredential; ImportError if azure-identity is missing.

Source code in src/pathlib_next/uri/__init__.py

def __init__(self, *uris: UriLike, **options):
    if self._raw_uris or self._initiated:
        return
    _uris: list[str | Uri] = []
    for uri in uris:
        if not uri:
            uri = ""
        if isinstance(uri, Uri):
            _uris.append(uri)
        elif isinstance(uri, (_pathlib.Path, Path)):
            try:
                uri = uri.as_uri()
            except ValueError:
                # as_uri() raises ValueError for a relative path, which
                # joins like a relative PurePath (see _RelativeLocalPath).
                uri = _RelativeLocalPath(_uriencode(uri.as_posix(), safe="/"))
            _uris.append(uri)
        elif isinstance(uri, (_pathlib.PurePath, Pathname)):
            _uris.append(_path_reference(uri.as_posix()))
        elif hasattr(uri, "as_uri"):
            path = uri.as_uri
            if callable(path):
                path = path()
            _uris.append(path)
        elif isinstance(uri, str):
            _uris.append(uri)
        elif isinstance(uri, bytes):
            _uris.append(uri.decode())
        else:
            path = None
            try:
                path = os.fspath(uri)
            except (TypeError, NotImplementedError):
                pass
            if not isinstance(path, str):
                raise TypeError(
                    "argument should be a str or an os.PathLike "
                    "object where __fspath__ returns a str, "
                    f"not {type(path).__name__!r}"
                )
            # Only __fspath__ is guaranteed here -- posix-normalize the
            # string itself rather than assuming an as_posix() method.
            posix = _pathlib.PurePath(path).as_posix()
            _uris.append(_path_reference(posix))
    self._raw_uris = _uris

account property

backend instance-attribute

container property

key property

rename(target)

Source code in src/pathlib_next/uri/schemes/az.py

def rename(self, target: "AzPath | Uri | str"):
    target = self._rename_target(target)
    # `with_path`, not `with_segments(target)`: the latter joined the Uri
    # object itself as a segment and raised TypeError for every str target.
    dest = target if isinstance(target, AzPath) else self.with_path(target.path)
    dest_key = dest.key
    if dest_key == self.key:
        # Copying a blob onto itself and then deleting the "source"
        # deletes the only copy. pathlib returns the new path.
        return dest
    if self._properties(self.key) is None:
        # No blob at the key: a prefix directory (stat() raises
        # FileNotFoundError when there is nothing at all). move() falls
        # back to copy + rm for it.
        self._pop_stat_hint()
        self.stat()
        raise NotImplementedError(f"rename() of the prefix directory {self}")
    source_blob_client = self._container.get_blob_client(self.key)
    source_url = source_blob_client.url
    dest_blob_client = self._container.get_blob_client(dest_key)
    with _translate_errors(self):
        # start_copy_from_url is async, poll for completion
        copy_props = dest_blob_client.start_copy_from_url(source_url)
        # Poll until copy is complete
        while _copy_status(copy_props) == "pending":
            import time

            time.sleep(0.1)
            dest_blob_client = self._container.get_blob_client(dest_key)
            copy_props = dest_blob_client.get_blob_properties()
        if _copy_status(copy_props) != "success":
            raise OSError(f"Copy failed: {self} -> {target}")
        source_blob_client.delete_blob()
    return dest

rm(recursive=False, missing_ok=False, ignore_error=False)

Source code in src/pathlib_next/uri/schemes/az.py

def rm(
    self,
    /,
    recursive=False,
    missing_ok=False,
    ignore_error: bool | _ty.Callable[[Exception, _ty.Self], bool] = False,
):
    if not recursive:
        return super().rm(
            recursive=recursive,
            missing_ok=missing_ok,
            ignore_error=ignore_error,
        )

    def on_error(error):
        if callable(ignore_error):
            return ignore_error(error, self)
        return bool(ignore_error)

    if not self.key:
        error = PermissionError("recursive container delete is not enabled")
        if not on_error(error):
            raise error
        return

    keys = []
    try:
        if self._properties(self.key) is not None:
            keys.append(self.key)
    except OSError as error:
        # Not "missing": deleting the prefix tree instead of the blob
        # would remove the wrong thing.
        if not on_error(error):
            raise
        return

    if not keys:
        marker = f"{self.key}/"
        try:
            with _translate_errors(self):
                keys.extend(
                    blob.name
                    for blob in self._container.list_blobs(name_starts_with=marker)
                )
        except Exception as error:
            if not on_error(error):
                raise
            return

    if not keys:
        if missing_ok:
            return
        error = FileNotFoundError(self)
        if not on_error(error):
            raise error
        return

    try:
        delete_blobs = getattr(self._container, "delete_blobs")
    except AttributeError:
        delete_blobs = None
    if callable(delete_blobs):
        for index in range(0, len(keys), 256):
            batch = keys[index : index + 256]
            try:
                delete_blobs(*batch)
            except Exception:
                # A partial failure has already deleted the other keys;
                # a rejected batch (auth on the batch endpoint, an
                # emulator without batch support) has deleted none.
                # Either way every key is retried on its own, so one
                # failing blob does not leave the rest behind.
                self._delete_each(batch, on_error, ignore_missing=True)
        return

    self._delete_each(keys, on_error)

rmdir()

Source code in src/pathlib_next/uri/schemes/az.py

def rmdir(self):
    if not self.stat().is_dir():
        raise NotADirectoryError(_errno.ENOTDIR, "Not a directory", str(self))
    marker = f"{self.key}/"
    count = 0
    with _translate_errors(self):
        for blob in self._container.list_blobs(name_starts_with=marker):
            if blob.name != marker:
                raise OSError(_errno.ENOTEMPTY, "Directory not empty", str(self))
            count += 1
            if count > 1:
                break
    marker_blob = self._container.get_blob_client(marker)
    try:
        with _translate_errors(self):
            marker_blob.delete_blob()
    except FileNotFoundError:
        pass

stat(*, follow_symlinks=True)

Source code in src/pathlib_next/uri/schemes/az.py

def stat(self, *, follow_symlinks=True):
    hint = self._pop_stat_hint()
    if hint is not None:
        return hint
    key = self.key
    if key == "":
        # A container root (or `az://account`, which holds containers):
        # ask the server, as s3: does with HeadBucket, so a mistyped
        # name does not read as an existing directory. A one-item
        # listing needs only the permission the rest of the path uses.
        with _translate_errors(self):
            if self.container:
                listing = self._container.list_blobs(
                    name_starts_with="", results_per_page=1
                )
            else:
                listing = self._client.list_containers(results_per_page=1)
            for _ in listing:
                break
        return FileStat(is_dir=True)
    # Only a not-found reply falls through to the prefix probe: a
    # transient or permission error read as "missing" let copy() replace
    # an existing blob.
    props = self._properties(key)
    if props is not None:
        return FileStat(
            st_size=props["size"],
            st_mtime=(
                int(props["last_modified"].timestamp())
                if props.get("last_modified")
                else 0
            ),
            is_dir=False,
        )
    # Not a blob at this exact key -- emulate a directory: any blob
    # under the "<key>/" prefix means this is a "directory".
    prefix = f"{key}/"
    with _translate_errors(self):
        for _ in self._container.list_blobs(name_starts_with=prefix):
            return FileStat(is_dir=True)
    raise FileNotFoundError(self)

unlink(missing_ok=False)

Source code in src/pathlib_next/uri/schemes/az.py

def unlink(self, missing_ok=False):
    try:
        st = self.stat()
    except FileNotFoundError:
        if missing_ok:
            return
        raise
    if st.is_dir():
        raise IsADirectoryError(_errno.EISDIR, "Is a directory", str(self))
    blob_client = self._container.get_blob_client(self.key)
    try:
        with _translate_errors(self):
            blob_client.delete_blob()
    except FileNotFoundError:
        # Deleted since the stat() above. Anything else -- a lease, a
        # permission error -- is not "already gone" and raises.
        if not missing_ok:
            raise