Source code for multistorageclient.types

# SPDX-FileCopyrightText: Copyright (c) 2024 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from __future__ import annotations

import posixpath
from abc import ABC, abstractmethod
from collections.abc import Iterator, Sequence
from dataclasses import asdict, dataclass, field, replace
from datetime import datetime, timezone
from enum import Enum
from typing import IO, Any, NamedTuple

from dateutil.parser import parse as dateutil_parser

MSC_PROTOCOL_NAME = "msc"
MSC_PROTOCOL = MSC_PROTOCOL_NAME + "://"

DEFAULT_RETRY_ATTEMPTS = 3
DEFAULT_RETRY_DELAY = 1.0
DEFAULT_RETRY_BACKOFF_MULTIPLIER = 2.0

# datetime.min is a naive datetime.
#
# This creates issues when doing datetime.astimezone(timezone.utc) since it assumes the local timezone for the naive datetime.
# If the local timezone is offset behind UTC, it attempts to subtract off the offset which goes below the representable limit (i.e. an underflow).
# A `ValueError: year 0 is out of range` is thrown as a result.
AWARE_DATETIME_MIN = datetime.min.replace(tzinfo=timezone.utc)




class SymlinkHandling(str, Enum):
    """Controls how symbolic links are handled during listing and sync operations.

    External symlinks (targets outside ``base_path``) are skipped by default in
    ``FOLLOW`` and ``PRESERVE``. Use ``FOLLOW_STRICT`` or ``PRESERVE_STRICT`` to
    fail instead.

    - ``FOLLOW``: Dereference internal symlinks; skip external symlinks. This is
      the default.
    - ``FOLLOW_STRICT``: Dereference internal symlinks; raise on external symlinks.
    - ``SKIP``: All symlinks are excluded from results.
    - ``PRESERVE``: Surface internal symlinks as leaf entries with
      :py:attr:`ObjectMetadata.symlink_target` populated; skip external symlinks.
      Directory symlinks are **not** recursed into.
    - ``PRESERVE_STRICT``: Same as ``PRESERVE`` for internal symlinks; raise on
      external symlinks.

    .. note::
       This option is only meaningful for POSIX file storage providers, which
       have native symlink semantics. Cloud storage providers (S3, GCS, Azure,
       OCI, AIS) ignore this parameter: they always list whatever is in the
       bucket/prefix and surface MSC's symlink convention (an empty object
       whose user metadata carries the target path) with
       :py:attr:`ObjectMetadata.symlink_target` populated -- object storage
       isn't a filesystem, so there is nothing to "follow" during listing.
    """

    FOLLOW = "follow"
    FOLLOW_STRICT = "follow_strict"
    SKIP = "skip"
    PRESERVE = "preserve"
    PRESERVE_STRICT = "preserve_strict"



# Maximum number of symlink hops allowed when resolving a symlink chain.
MAX_SYMLINK_DEPTH = 8




class SignerType(str, Enum):
    """Supported signer backends for presigned URL generation."""

    S3 = "s3"
    CLOUDFRONT = "cloudfront"
    AZURE = "azure"





@dataclass
class Credentials:
    """
    A data class representing the credentials needed to access a storage provider.
    """

    #: The access key for authentication.
    access_key: str
    #: The secret key for authentication.
    secret_key: str
    #: An optional security token for temporary credentials.
    token: str | None
    #: The expiration time of the credentials in ISO 8601 format.
    expiration: str | None
    #: A dictionary for storing custom key-value pairs.
    custom_fields: dict[str, Any] = field(default_factory=dict)



    def is_expired(self) -> bool:
        """
        Checks if the credentials are expired based on the expiration time.

        :return: ``True`` if the credentials are expired, ``False`` otherwise.
        """
        expiry = dateutil_parser(self.expiration) if self.expiration else None
        if expiry is None:
            return False
        return expiry <= datetime.now(tz=timezone.utc)




    def get_custom_field(self, key: str, default: Any = None) -> Any:
        """
        Retrieves a value from custom fields by its key.

        :param key: The key to look up in custom fields.
        :param default: The default value to return if the key is not found.
        :return: The value associated with the key, or the default value if not found.
        """
        return self.custom_fields.get(key, default)






@dataclass
class ObjectMetadata:
    """
    A data class that represents the metadata associated with an object stored in a cloud storage service. This metadata
    includes both required and optional information about the object.
    """

    #: Relative path of the object.
    key: str
    #: The size of the object in bytes.
    content_length: int
    #: The timestamp indicating when the object was last modified.
    last_modified: datetime
    type: str = "file"
    #: The MIME type of the object.
    content_type: str | None = field(default=None)
    #: The entity tag (ETag) of the object.
    etag: str | None = field(default=None)
    #: The storage class of the object.
    storage_class: str | None = field(default=None)

    metadata: dict[str, Any] | None = field(default=None)

    #: Symlink target relative to the symlink's own parent directory
    #: (e.g. ``"../target.txt"``), or ``None`` for non-symlink entries.
    #: See :meth:`encode_symlink_target` / :meth:`resolve_symlink_target`.
    symlink_target: str | None = field(default=None)



    @staticmethod
    def from_dict(data: dict) -> ObjectMetadata:
        """
        Creates an ObjectMetadata instance from a dictionary (parsed from JSON).
        """
        try:
            last_modified = dateutil_parser(data["last_modified"])
            key = data.get("key")
            if key is None:
                raise ValueError("Missing required field: 'key'")
            return ObjectMetadata(
                key=key,
                content_length=data["content_length"],
                last_modified=last_modified,
                type=data.get("type", "file"),  # default to file
                content_type=data.get("content_type"),
                etag=data.get("etag"),
                storage_class=data.get("storage_class"),
                metadata=data.get("metadata"),
                symlink_target=data.get("symlink_target"),
            )
        except KeyError as e:
            raise ValueError("Missing required field.") from e




    def replace(self, **changes: Any) -> ObjectMetadata:
        """Return a shallow copy of this object with the given fields overridden."""
        return replace(self, **changes)




    def to_dict(self) -> dict:
        data = asdict(self)
        data["last_modified"] = self.last_modified.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%fZ")
        return {k: v for k, v in data.items() if v is not None}




    @staticmethod
    def encode_symlink_target(symlink_key: str, target: str) -> str:
        """Encode a logical ``target`` into the canonical
        :py:attr:`symlink_target` storage form: the target expressed relative
        to the symlink's own parent directory.

        This is the single format every backend persists — native OS symlinks
        on POSIX, ``msc-symlink-target`` user metadata on object stores, and
        manifest entries on metadata-provider-backed stores — so callers
        holding a logical key can delegate the ``posixpath.relpath`` dance to
        this helper.

        :param symlink_key: Key/path of the symlink itself.
        :param target: Target expressed as a logical key in the same frame as
            ``symlink_key`` (i.e. both rooted at the same base).
        :return: ``target`` re-expressed relative to
            ``posixpath.dirname(symlink_key)``. When ``symlink_key`` has no
            parent, ``target`` is returned unchanged.
        """
        parent = posixpath.dirname(symlink_key)
        return posixpath.relpath(target, parent) if parent else target




    @staticmethod
    def resolve_symlink_target(symlink_key: str, symlink_target: str) -> str:
        """Resolve a stored :py:attr:`symlink_target` to the logical key it
        points to.

        Inverse of :meth:`encode_symlink_target`: joins the parent-relative
        ``symlink_target`` with the symlink's own parent directory and
        normalises the result.

        :param symlink_key: Key/path of the symlink itself.
        :param symlink_target: Value read from
            :py:attr:`ObjectMetadata.symlink_target`.
        :return: Logical key the symlink points to, in the same frame as
            ``symlink_key``.
        """
        return posixpath.normpath(posixpath.join(posixpath.dirname(symlink_key), symlink_target))






class CredentialsProvider(ABC):
    """
    Abstract base class for providing credentials to access a storage provider.
    """



    @abstractmethod
    def get_credentials(self) -> Credentials:
        """
        Retrieves the current credentials.

        :return: The current credentials used for authentication.
        """




    @abstractmethod
    def refresh_credentials(self) -> None:
        """
        Refreshes the credentials if they are expired or about to expire.
        """






@dataclass
class Range:
    """
    A data class that represents a byte range for read operations.
    """

    #: The start offset in bytes.
    offset: int
    #: The number of bytes to read.
    size: int





class StorageProvider(ABC):
    """
    Abstract base class for interacting with a storage provider.
    """



    @abstractmethod
    def put_object(
        self,
        path: str,
        body: bytes,
        if_match: str | None = None,
        if_none_match: str | None = None,
        attributes: dict[str, Any] | None = None,
    ) -> None:
        """
        Uploads an object to the storage provider.

        :param path: The path where the object will be stored.
        :param body: The content of the object to store.
        :param if_match: Optional If-Match value for conditional upload.
        :param if_none_match: Optional If-None-Match value for conditional upload.
        :param attributes: The attributes to add to the file.
        """




    @abstractmethod
    def get_object(self, path: str, byte_range: Range | None = None) -> bytes:
        """
        Retrieves an object from the storage provider.

        :param path: The path where the object is stored.
        :param byte_range: Optional byte range (offset, length) to read.
        :return: The content of the retrieved object.
        """




    @abstractmethod
    def copy_object(self, src_path: str, dest_path: str) -> None:
        """
        Copies an object from source to destination in the storage provider.

        :param src_path: The path of the source object to copy.
        :param dest_path: The path of the destination.
        """




    @abstractmethod
    def delete_object(self, path: str, if_match: str | None = None) -> None:
        """
        Deletes an object from the storage provider.

        :param path: The path of the object to delete.
        :param if_match: Optional if-match value to use for conditional deletion.
        """




    @abstractmethod
    def delete_objects(self, paths: list[str]) -> None:
        """
        Deletes multiple objects from the storage provider.

        :param paths: A list of paths of objects to delete.
        """




    @abstractmethod
    def make_symlink(self, path: str, target: str) -> None:
        """
        Creates a symbolic link at ``path`` pointing to ``target``.

        On POSIX backends this creates a native OS symlink with a relative target path.
        On object-store backends this creates a zero-byte marker object with the target
        stored in user metadata (``msc-symlink-target``).

        :param path: The path where the symlink will be created.
        :param target: The logical key that the symlink points to.
        """




    @abstractmethod
    def get_object_metadata(self, path: str, strict: bool = True) -> ObjectMetadata:
        """
        Retrieves metadata or information about an object stored in the provider.

        :param path: The path of the object.
        :param strict: When ``True``, performs additional validation to determine whether the path refers to a directory.

        :return: A metadata object containing the information about the object.
        """




    @abstractmethod
    def list_objects(
        self,
        path: str,
        start_after: str | None = None,
        end_at: str | None = None,
        include_directories: bool = False,
        attribute_filter_expression: str | None = None,
        show_attributes: bool = False,
        symlink_handling: SymlinkHandling = SymlinkHandling.FOLLOW,
    ) -> Iterator[ObjectMetadata]:
        """
        Lists objects in the storage provider under the specified path.

        :param path: The path to list objects under. The path must be a valid file or subdirectory path, cannot be partial or just "prefix".
        :param start_after: The key to start after (i.e. exclusive). An object with this key doesn't have to exist.
        :param end_at: The key to end at (i.e. inclusive). An object with this key doesn't have to exist.
        :param include_directories: Whether to include directories in the result. When ``True``, directories are returned alongside objects.
        :param attribute_filter_expression: The attribute filter expression to apply to the result.
        :param show_attributes: Whether to return attributes in the result.  There will be performance impact if this is True as now we need to get object metadata for each object.
        :param symlink_handling: How to handle symbolic links during listing.

        :return: An iterator over objects metadata under the specified path.
        """




    @abstractmethod
    def list_objects_recursive(
        self,
        path: str = "",
        start_after: str | None = None,
        end_at: str | None = None,
        max_workers: int = 32,
        look_ahead: int = 2,
        symlink_handling: SymlinkHandling = SymlinkHandling.FOLLOW,
    ) -> Iterator[ObjectMetadata]:
        """
        Lists files recursively in the storage provider under the specified path.

        :param path: The path to list objects under.
        :param start_after: The key to start after (i.e. exclusive). An object with this key doesn't have to exist.
        :param end_at: The key to end at (i.e. inclusive). An object with this key doesn't have to exist.
        :param max_workers: Maximum concurrent workers for provider-level recursive listing.
        :param look_ahead: Prefixes to buffer per worker for provider-level recursive listing.
        :param symlink_handling: How to handle symbolic links during listing.
        :return: An iterator over object metadata under the specified path.
        """




    @abstractmethod
    def upload_file(self, remote_path: str, f: str | IO, attributes: dict[str, Any] | None = None) -> None:
        """
        Uploads a file from the local file system to the storage provider.

        :param remote_path: The path where the object will be stored.
        :param f: The source file to upload. This can either be a string representing the local
            file path, or a file-like object (e.g., an open file handle).
        :param attributes: The attributes to add to the file if a new file is created.
        """




    @abstractmethod
    def download_file(self, remote_path: str, f: str | IO, metadata: ObjectMetadata | None = None) -> None:
        """
        Downloads a file from the storage provider to the local file system.

        :param remote_path: The path of the file to download.
        :param f: The destination for the downloaded file. This can either be a string representing
            the local file path where the file will be saved, or a file-like object to write the
            downloaded content into.
        :param metadata: Metadata about the object to download.
        """




    @abstractmethod
    def download_files(
        self,
        remote_paths: list[str],
        local_paths: list[str],
        metadata: Sequence[ObjectMetadata | None] | None = None,
        max_workers: int = 16,
    ) -> None:
        """
        Downloads multiple files from the storage provider to the local file system.

        :param remote_paths: List of remote paths of files to download.
        :param local_paths: List of local file paths to save the downloaded files to.
        :param metadata: Optional per-file metadata used to decide between regular and multipart download.
        :param max_workers: Maximum number of concurrent download workers (default: 16).
        :raises ValueError: If remote_paths and local_paths have different lengths.
        """




    @abstractmethod
    def upload_files(
        self,
        local_paths: list[str],
        remote_paths: list[str],
        attributes: Sequence[dict[str, Any] | None] | None = None,
        max_workers: int = 16,
    ) -> None:
        """
        Uploads multiple files from the local file system to the storage provider.

        :param local_paths: List of local file paths to upload.
        :param remote_paths: List of remote paths to upload the files to.
        :param attributes: Optional list of per-file attributes to add. When provided, must have the same length
            as local_paths/remote_paths. Each element may be ``None`` for files that need no attributes.
        :param max_workers: Maximum number of concurrent upload workers (default: 16).
        :raises ValueError: If local_paths and remote_paths have different lengths.
        :raises ValueError: If attributes is provided and has a different length than local_paths.
        """




    @abstractmethod
    def glob(self, pattern: str, attribute_filter_expression: str | None = None) -> list[str]:
        """
        Matches and retrieves a list of object keys in the storage provider that match the specified pattern.

        :param pattern: The pattern to match object keys against, supporting wildcards (e.g., ``*.txt``).
        :param attribute_filter_expression: The attribute filter expression to apply to the result.

        :return: A list of object keys that match the specified pattern.
        """




    @abstractmethod
    def is_file(self, path: str) -> bool:
        """
        Checks whether the specified key in the storage provider points to a file (as opposed to a folder or directory).

        :param path: The path to check.

        :return: ``True`` if the key points to a file, ``False`` if it points to a directory or folder.
        """




    def generate_presigned_url(
        self,
        path: str,
        *,
        method: str = "GET",
        signer_type: SignerType | None = None,
        signer_options: dict[str, Any] | None = None,
    ) -> str:
        """
        Generate a pre-signed URL granting temporary access to the object at *path*.

        :param path: The object path within the storage provider.
        :param method: The HTTP method the URL should authorise (e.g. ``"GET"``, ``"PUT"``).
        :param signer_type: The signing backend to use. ``None`` means the provider's native signer.
        :param signer_options: Backend-specific options forwarded to the signer.
        :return: A pre-signed URL string.
        :raises NotImplementedError: If this storage provider does not support presigned URLs.
        """
        raise NotImplementedError(f"{type(self).__name__} does not support presigned URL generation.")






class ResolvedPathState(str, Enum):
    """
    Enum representing the state of a resolved path.
    """

    EXISTS = "exists"  # File currently exists
    DELETED = "deleted"  # File existed before but has been deleted
    UNTRACKED = "untracked"  # File never existed or was never tracked





class ResolvedPath(NamedTuple):
    """
    Result of resolving a virtual path to a physical path.

    :param physical_path: The physical path in storage backend
    :param state: The state of the path (EXISTS, DELETED, or UNTRACKED)
    :param profile: Optional profile name for routing in CompositeStorageClient.
        None means use current client's storage provider.
        String means route to named child StorageClient.

    State meanings:
    - EXISTS: File currently exists in metadata
    - DELETED: File existed before but has been deleted (soft delete)
    - UNTRACKED: File never existed or was never tracked
    """

    physical_path: str
    state: ResolvedPathState
    profile: str | None = None

    @property
    def exists(self) -> bool:
        """Backward compatibility property: True if state is EXISTS."""
        return self.state == ResolvedPathState.EXISTS





class MetadataProvider(ABC):
    """
    Abstract base class for accessing file metadata.
    """



    @abstractmethod
    def list_objects(
        self,
        path: str,
        start_after: str | None = None,
        end_at: str | None = None,
        include_directories: bool = False,
        attribute_filter_expression: str | None = None,
        show_attributes: bool = False,
    ) -> Iterator[ObjectMetadata]:
        """
        Lists objects in the metadata provider under the specified path.

        :param path: The path to list objects under. The path must be a valid file or subdirectory path, cannot be partial or just "prefix".
        :param start_after: The key to start after (i.e. exclusive). An object with this key doesn't have to exist.
        :param end_at: The key to end at (i.e. inclusive). An object with this key doesn't have to exist.
        :param include_directories: Whether to include directories in the result. When ``True``, directories are returned alongside objects.
        :param attribute_filter_expression: The attribute filter expression to apply to the result.
        :param show_attributes: Whether to return attributes in the result. Depending on implementation, there may be a performance impact if this is set to ``True``.

        :return: An iterator over object metadata under the specified path.
        """




    @abstractmethod
    def get_object_metadata(self, path: str, include_pending: bool = False) -> ObjectMetadata:
        """
        Retrieves metadata or information about an object or directory stored in the provider.

        If the path does not match a file, implementations should check whether the path
        represents a valid directory (i.e. files exist under the path prefix) and return
        directory metadata accordingly.

        :param path: The path of the object or directory.
        :param include_pending: Whether to include metadata that is not yet committed.

        :return: A metadata object containing the information about the object or directory.
        :raises FileNotFoundError: If no object or directory exists at the specified path.
        """




    @abstractmethod
    def glob(self, pattern: str, attribute_filter_expression: str | None = None) -> list[str]:
        """
        Matches and retrieves a list of object keys in the storage provider that match the specified pattern.

        :param pattern: The pattern to match object keys against, supporting wildcards (e.g., ``*.txt``).
        :param attribute_filter_expression: The attribute filter expression to apply to the result.

        :return: A list of object keys that match the specified pattern.
        """




    @abstractmethod
    def realpath(self, logical_path: str) -> ResolvedPath:
        """
        Resolves a logical path to its physical storage path.

        This method checks if the object exists in the committed state and returns
        the appropriate physical path with the current state of the path.

        :param logical_path: The user-facing logical path

        :return: ResolvedPath with physical_path and state:
            - ResolvedPathState.EXISTS: File currently exists
            - ResolvedPathState.UNTRACKED: File never existed
            - ResolvedPathState.DELETED: File was deleted
            If state is EXISTS, physical_path is the committed storage path.
            Otherwise, physical_path is typically the logical_path as fallback.
        """




    @abstractmethod
    def generate_physical_path(self, logical_path: str, for_overwrite: bool = False) -> ResolvedPath:
        """
        Generates a physical storage path for writing a new or overwritten object.

        This method is used for write operations to determine where the object should
        be physically stored. Implementations can use this to:
        - Generate UUID-based paths for deduplication
        - Create versioned paths (file-v1.txt, file-v2.txt) for time travel
        - Implement path rewriting strategies

        :param logical_path: The user-facing logical path
        :param for_overwrite: When ``True``, indicates the path is for overwriting an existing object.
            Implementations may generate unique paths for overwrites to support versioning.

        :return: ResolvedPath with physical_path for writing. The exists flag indicates
            whether the logical path currently exists in committed state (for overwrite scenarios).
        """




    @abstractmethod
    def add_file(self, path: str, metadata: ObjectMetadata) -> None:
        """
        Add a file to be tracked by the :py:class:`MetadataProvider`. Does not have to be
        reflected in listing until a :py:meth:`MetadataProvider.commit_updates` forces a persist.
        This function must tolerate duplicate calls (idempotent behavior).

        :param path: User-supplied virtual path
        :param metadata: physical file metadata from StorageProvider
        """




    @abstractmethod
    def remove_file(self, path: str) -> None:
        """
        Remove a file tracked by the :py:class:`MetadataProvider`. Does not have to be
        reflected in listing until a :py:meth:`MetadataProvider.commit_updates` forces a persist.
        This function must tolerate duplicate calls (idempotent behavior).

        :param path: User-supplied virtual path
        """




    @abstractmethod
    def commit_updates(self) -> None:
        """
        Commit any newly adding files, used in conjunction with :py:meth:`MetadataProvider.add_file`.
        :py:class:`MetadataProvider` will persistently record any metadata changes.
        """




    @abstractmethod
    def is_writable(self) -> bool:
        """
        Returns ``True`` if the :py:class:`MetadataProvider` supports writes else ``False``.
        """




    @abstractmethod
    def allow_overwrites(self) -> bool:
        """
        Returns ``True`` if the :py:class:`MetadataProvider` allows overwriting existing files else ``False``.
        When ``True``, :py:meth:`add_file` will not raise an error if the file already exists.
        """




    @abstractmethod
    def should_use_soft_delete(self) -> bool:
        """
        Returns ``True`` if the :py:class:`MetadataProvider` should use soft-delete behavior else ``False``.

        When ``True``, delete operations will only mark files as deleted in metadata without removing
        the physical data from storage. The file will return :py:class:`ResolvedPathState.DELETED` state
        when queried and will not appear in listings.

        When ``False``, delete operations will remove both the metadata and the physical file from storage
        (hard delete).
        """






@dataclass
class StorageProviderConfig:
    """
    A data class that represents the configuration needed to initialize a storage provider.
    """

    #: The name or type of the storage provider (e.g., ``s3``, ``gcs``, ``oci``, ``azure``).
    type: str
    #: Additional options required to configure the storage provider (e.g., endpoint URLs, region, etc.).
    options: dict[str, Any] | None = None





@dataclass
class StorageBackend:
    """
    Represents configuration for a single storage backend.
    """

    storage_provider_config: StorageProviderConfig
    credentials_provider: CredentialsProvider | None = None
    replicas: list[Replica] = field(default_factory=list)





class ProviderBundle(ABC):
    """
    Abstract base class that serves as a container for various providers (storage, credentials, and metadata)
    that interact with a storage service. The :py:class:`ProviderBundle` abstracts access to these providers, allowing for
    flexible implementations of cloud storage solutions.
    """

    @property
    @abstractmethod
    def storage_provider_config(self) -> StorageProviderConfig:
        """
        :return: The configuration for the storage provider, which includes the provider
                    name/type and additional options.
        """

    @property
    @abstractmethod
    def credentials_provider(self) -> CredentialsProvider | None:
        """
        :return: The credentials provider responsible for managing authentication credentials
                    required to access the storage service.
        """

    @property
    @abstractmethod
    def metadata_provider(self) -> MetadataProvider | None:
        """
        :return: The metadata provider responsible for retrieving metadata about objects in the storage service.
        """

    @property
    @abstractmethod
    def replicas(self) -> list[Replica]:
        """
        :return: The replicas configuration for this provider bundle, if any.
        """





class ProviderBundleV2(ABC):
    """
    Abstract base class that serves as a container for various providers (storage, credentials, and metadata)
    that interact with one or multiple storage service. The :py:class:`ProviderBundleV2` abstracts access to these providers, allowing for
    flexible implementations of cloud storage solutions.

    """

    @property
    @abstractmethod
    def storage_backends(self) -> dict[str, StorageBackend]:
        """
        :return: Mapping of storage backend name -> StorageBackend. Must have at least one backend.
        """

    @property
    @abstractmethod
    def metadata_provider(self) -> MetadataProvider | None:
        """
        :return: The metadata provider responsible for retrieving metadata about objects in the storage service. If there are multiple backends, this is required.
        """





@dataclass
class RetryConfig:
    """
    A data class that represents the configuration for retry strategy.
    """

    #: The number of attempts before giving up. Must be at least 1.
    attempts: int = DEFAULT_RETRY_ATTEMPTS
    #: The base delay (in seconds) for exponential backoff. Must be a non-negative value.
    delay: float = DEFAULT_RETRY_DELAY
    #: The backoff multiplier for exponential backoff. Must be at least 1.0.
    backoff_multiplier: float = DEFAULT_RETRY_BACKOFF_MULTIPLIER

    def __post_init__(self) -> None:
        if self.attempts < 1:
            raise ValueError("Attempts must be at least 1.")
        if self.delay < 0:
            raise ValueError("Delay must be a non-negative number.")
        if self.backoff_multiplier < 1.0:
            raise ValueError("Backoff multiplier must be at least 1.0.")





class RetryableError(Exception):
    """
    Exception raised for errors that should trigger a retry.
    """





@dataclass(frozen=True)
class BatchTransferFailure:
    """
    A failed item from a batch upload or download operation.
    """

    #: The item index in the batch request that failed.
    index: int
    #: The source path for the failed transfer.
    source_path: str
    #: The destination path for the failed transfer.
    destination_path: str
    #: The underlying exception raised for this item.
    error: Exception





class BatchTransferError(Exception):
    """
    Exception raised when one or more items fail in a batch transfer operation.
    """

    def __init__(self, failures: Sequence[BatchTransferFailure]):
        if not failures:
            raise ValueError("BatchTransferError requires at least one failure.")
        self.failures = list(failures)
        super().__init__(self._format_message())

    def _format_message(self) -> str:
        details = ", ".join(
            f"index {failure.index} ({type(failure.error).__name__}: {failure.error})" for failure in self.failures[:3]
        )
        if len(self.failures) > 3:
            details += f", and {len(self.failures) - 3} more"
        return f"{len(self.failures)} batch transfer item(s) failed: {details}"





class PreconditionFailedError(Exception):
    """
    Exception raised when a precondition fails. e.g. if-match, if-none-match, etc.
    """





class NotModifiedError(Exception):
    """
    Raised when a conditional operation fails because the resource has not been modified.

    This typically occurs when using if-none-match with a specific generation/etag
    and the resource's current generation/etag matches the specified one.
    """





class SourceVersionCheckMode(Enum):
    """
    Enum for controlling source version checking behavior.
    """

    INHERIT = "inherit"  # Inherit from configuration (cache config)
    ENABLE = "enable"  # Always check source version
    DISABLE = "disable"  # Never check source version





@dataclass
class Replica:
    """
    A tier of storage that can be used to store data.
    """

    replica_profile: str
    read_priority: int





class AutoCommitConfig:
    """
    A data class that represents the configuration for auto commit.
    """

    interval_minutes: float | None  # The interval in minutes for auto commit.
    at_exit: bool = False  # if True, commit on program exit

    def __init__(self, interval_minutes: float | None = None, at_exit: bool = False) -> None:
        self.interval_minutes = interval_minutes
        self.at_exit = at_exit





class ExecutionMode(Enum):
    """
    Enum for controlling execution mode in sync operations.
    """

    LOCAL = "local"
    RAY = "ray"





class PatternType(Enum):
    """
    Type of pattern operation for include/exclude filtering.
    """

    INCLUDE = "include"
    EXCLUDE = "exclude"



# Type alias for pattern matching
PatternList = list[tuple[PatternType, str]]




@dataclass
class DryrunResult:
    """
    Holds references to JSONL files produced by a dryrun sync operation.

    Each file contains one JSON object per line, matching the :py:class:`ObjectMetadata`
    serialization format (see :py:meth:`ObjectMetadata.to_dict` / :py:meth:`ObjectMetadata.from_dict`).

    The caller is responsible for cleaning up the files when they are no longer needed.
    """

    #: Path to a JSONL file listing source objects that would be added to the target.
    files_to_add: str
    #: Path to a JSONL file listing target objects that would be deleted.
    files_to_delete: str





@dataclass
class SyncResult:
    """
    A data class that represents the summary of a sync operation.
    """

    #: The total number of work units tracked for progress (including files from both source and target after filtering). Each work unit represents an ADD or DELETE operation.
    total_work_units: int = 0
    #: The total number of files processed to the target.
    total_files_added: int = 0
    #: The total number of files deleted from the target.
    total_files_deleted: int = 0
    #: The total number of bytes transferred to the target.
    total_bytes_added: int = 0
    #: The total number of bytes deleted from the target.
    total_bytes_deleted: int = 0
    #: The total time taken to process the sync operation.
    total_time_seconds: float = 0.0
    #: Dryrun details with paths to JSONL files. ``None`` for normal (non-dryrun) sync operations.
    dryrun: DryrunResult | None = None

    def __str__(self) -> str:
        header = "Sync dryrun statistics:" if self.dryrun else "Sync statistics:"
        lines = (
            f"{header}\n"
            f"  Work units: {self.total_work_units}\n"
            f"  Files added: {self.total_files_added}\n"
            f"  Files deleted: {self.total_files_deleted}\n"
            f"  Bytes added: {self.total_bytes_added}\n"
            f"  Bytes deleted: {self.total_bytes_deleted}\n"
            f"  Time elapsed: {self.total_time_seconds:.2f}s"
        )
        if self.dryrun:
            lines += f"\n  Files to add: {self.dryrun.files_to_add}\n  Files to delete: {self.dryrun.files_to_delete}"
        return lines





class SyncError(RuntimeError):
    """
    Exception raised when errors occur during a sync operation.

    This exception includes the partial SyncResult showing what was accomplished
    before the error occurred, allowing users to understand the state of the sync.

    :param message: The error message describing what went wrong.
    :param sync_result: The partial SyncResult with statistics from the failed sync operation.
    """

    def __init__(self, message: str, sync_result: SyncResult):
        super().__init__(message)
        self.sync_result = sync_result

    def __str__(self) -> str:
        sync_stats = str(self.sync_result).replace("Sync statistics:", "Partial sync statistics:")
        return f"{super().__str__()}\n\n{sync_stats}"