Source code for multistorageclient.config

   1# SPDX-FileCopyrightText: Copyright (c) 2024 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
   2# SPDX-License-Identifier: Apache-2.0
   3#
   4# Licensed under the Apache License, Version 2.0 (the "License");
   5# you may not use this file except in compliance with the License.
   6# You may obtain a copy of the License at
   7#
   8# http://www.apache.org/licenses/LICENSE-2.0
   9#
  10# Unless required by applicable law or agreed to in writing, software
  11# distributed under the License is distributed on an "AS IS" BASIS,
  12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13# See the License for the specific language governing permissions and
  14# limitations under the License.
  15
  16import copy
  17import json
  18import logging
  19import os
  20import tempfile
  21from collections import defaultdict
  22from collections.abc import Callable, Iterable
  23from pathlib import Path
  24from typing import Any
  25from urllib.parse import urlparse
  26
  27import yaml
  28
  29from .cache import DEFAULT_CACHE_LINE_SIZE, DEFAULT_CACHE_SIZE, CacheManager
  30from .caching.cache_config import CacheConfig, EvictionPolicyConfig
  31from .providers.manifest_metadata import ManifestMetadataProvider
  32from .rclone import read_rclone_config
  33from .schema import validate_config
  34from .telemetry import Telemetry
  35from .telemetry import init as telemetry_init
  36from .types import (
  37    DEFAULT_RETRY_ATTEMPTS,
  38    DEFAULT_RETRY_BACKOFF_MULTIPLIER,
  39    DEFAULT_RETRY_DELAY,
  40    MSC_PROTOCOL,
  41    AutoCommitConfig,
  42    CredentialsProvider,
  43    MetadataProvider,
  44    ProviderBundle,
  45    ProviderBundleV2,
  46    Replica,
  47    RetryConfig,
  48    StorageBackend,
  49    StorageProvider,
  50    StorageProviderConfig,
  51)
  52from .utils import build_full_path, expand_env_vars, import_class, merge_dictionaries_no_overwrite
  53
  54# Constants related to implicit profiles
  55SUPPORTED_IMPLICIT_PROFILE_PROTOCOLS = ("s3", "gs", "ais", "file")
  56PROTOCOL_TO_PROVIDER_TYPE_MAPPING = {
  57    "s3": "s3",
  58    "gs": "gcs",
  59    "ais": "ais",
  60    "file": "file",
  61}
  62
  63
  64# Template for creating implicit profile configurations
  65def create_implicit_profile_config(profile_name: str, protocol: str, base_path: str) -> dict:
  66    """
  67    Create a configuration dictionary for an implicit profile.
  68
  69    :param profile_name: The name of the profile (e.g., "_s3-bucket1")
  70    :param protocol: The storage protocol (e.g., "s3", "gs", "ais")
  71    :param base_path: The base path (e.g., bucket name) for the storage provider
  72
  73    :return: A configuration dictionary for the implicit profile
  74    """
  75    provider_type = PROTOCOL_TO_PROVIDER_TYPE_MAPPING[protocol]
  76    return {
  77        "profiles": {profile_name: {"storage_provider": {"type": provider_type, "options": {"base_path": base_path}}}}
  78    }
  79
  80
  81RESERVED_POSIX_PROFILE_NAME = "__filesystem__"
  82DEFAULT_POSIX_PROFILE = create_implicit_profile_config(RESERVED_POSIX_PROFILE_NAME, "file", "/")
  83
  84STORAGE_PROVIDER_MAPPING = {
  85    "file": "PosixFileStorageProvider",
  86    "s3": "S3StorageProvider",
  87    "gcs": "GoogleStorageProvider",
  88    "oci": "OracleStorageProvider",
  89    "azure": "AzureBlobStorageProvider",
  90    "ais": "AIStoreStorageProvider",
  91    "ais_s3": "AIStoreS3StorageProvider",
  92    "s8k": "S8KStorageProvider",
  93    "gcs_s3": "GoogleS3StorageProvider",
  94    "s3_cuobject": "S3CuObjectStorageProvider",
  95    "huggingface": "HuggingFaceStorageProvider",
  96}
  97
  98CREDENTIALS_PROVIDER_MAPPING = {
  99    "S3Credentials": "StaticS3CredentialsProvider",
 100    "AzureCredentials": "StaticAzureCredentialsProvider",
 101    "DefaultAzureCredentials": "DefaultAzureCredentialsProvider",
 102    "AISCredentials": "StaticAISCredentialProvider",
 103    "GoogleIdentityPoolCredentialsProvider": "GoogleIdentityPoolCredentialsProvider",
 104    "GoogleServiceAccountCredentialsProvider": "GoogleServiceAccountCredentialsProvider",
 105    "HuggingFaceCredentials": "HuggingFaceCredentialsProvider",
 106    "FileBasedCredentials": "FileBasedCredentialsProvider",
 107}
 108
 109
 110def _resolve_include_path(include_path: str, parent_config_path: str) -> str:
 111    """
 112    Resolve include path (absolute or relative to parent config file).
 113
 114    :param include_path: Path from include keyword (can be absolute or relative)
 115    :param parent_config_path: Absolute path of the config file containing the include
 116    :return: Absolute, normalized path to the included config file
 117    """
 118    if os.path.isabs(include_path):
 119        return os.path.abspath(include_path)
 120    else:
 121        parent_dir = os.path.dirname(parent_config_path)
 122        return os.path.abspath(os.path.join(parent_dir, include_path))
 123
 124
 125def _merge_profiles(
 126    base_profiles: dict[str, Any],
 127    new_profiles: dict[str, Any],
 128    base_path: str,
 129    new_path: str,
 130) -> dict[str, Any]:
 131    """
 132    Merge profiles from two config files with conflict detection.
 133
 134    Profiles with the same name are allowed if their definitions are identical (idempotent).
 135    If a profile name exists in both configs with different definitions, an error is raised.
 136
 137    NOTE: This function performs SHALLOW merging - each profile is treated as a complete unit.
 138    We cannot use merge_dictionaries_no_overwrite() here because it would recursively merge
 139    profile fields, which could lead to unintended behavior. For example:
 140
 141    base_config:
 142        profiles:
 143            my-profile:
 144                storage_provider:
 145                    type: s3
 146                    options:
 147                        base_path: bucket
 148
 149    new_config:
 150        profiles:
 151            my-profile:
 152                credentials_provider:
 153                    type: S3Credentials
 154                    options:
 155                        access_key: foo
 156                        secret_key: bar
 157
 158    If we used merge_dictionaries_no_overwrite(), the result would be a profile with BOTH storage_provider
 159    and credentials_provider, which might not be the intended configuration.
 160
 161    :param base_profiles: Profiles from the base config
 162    :param new_profiles: Profiles from the new config to merge
 163    :param base_path: Path to the base config file (for error messages)
 164    :param new_path: Path to the new config file (for error messages)
 165    :return: Merged profiles dictionary
 166    :raises ValueError: If any profile name exists in both configs with different definitions
 167    """
 168    result = base_profiles.copy()
 169    conflicting_profiles = []
 170
 171    for profile_name, new_profile_def in new_profiles.items():
 172        if profile_name in result:
 173            if result[profile_name] != new_profile_def:
 174                conflicting_profiles.append(profile_name)
 175        else:
 176            result[profile_name] = new_profile_def
 177
 178    if conflicting_profiles:
 179        conflicts_list = ", ".join(f"'{p}'" for p in sorted(conflicting_profiles))
 180        raise ValueError(f"Profile conflict: {conflicts_list} defined differently in {base_path} and {new_path}")
 181
 182    return result
 183
 184
 185def _merge_opentelemetry(
 186    base_otel: dict[str, Any],
 187    new_otel: dict[str, Any],
 188    base_path: str,
 189    new_path: str,
 190) -> dict[str, Any]:
 191    """
 192    Merge opentelemetry configurations with special handling for metrics.attributes.
 193
 194    Merge strategy:
 195    - metrics.attributes: Concatenate arrays (preserve order for fallback mechanism)
 196    - Other fields: Use idempotent check (must be identical)
 197
 198    Attributes providers are concatenated to support fallback scenarios where multiple
 199    providers of the same type supply the same attribute key with different sources.
 200    The order matters - later providers override earlier ones.
 201
 202    :param base_otel: Base opentelemetry config
 203    :param new_otel: New opentelemetry config to merge
 204    :param base_path: Path to base config file (for error messages)
 205    :param new_path: Path to new config file (for error messages)
 206    :return: Merged opentelemetry configuration
 207    :raises ValueError: If conflicts are detected
 208    """
 209    # Merge metrics.attributes
 210    base_attrs = base_otel.get("metrics", {}).get("attributes", [])
 211    new_attrs = new_otel.get("metrics", {}).get("attributes", [])
 212    merged_attrs = base_attrs + new_attrs
 213
 214    # Merge other opentelemetry fields
 215    base_otel_without_attrs = copy.deepcopy(base_otel)
 216    if "metrics" in base_otel_without_attrs and "attributes" in base_otel_without_attrs["metrics"]:
 217        del base_otel_without_attrs["metrics"]["attributes"]
 218
 219    new_otel_without_attrs = copy.deepcopy(new_otel)
 220    if "metrics" in new_otel_without_attrs and "attributes" in new_otel_without_attrs["metrics"]:
 221        del new_otel_without_attrs["metrics"]["attributes"]
 222
 223    merged, conflicts = merge_dictionaries_no_overwrite(
 224        base_otel_without_attrs, new_otel_without_attrs, allow_idempotent=True
 225    )
 226
 227    if conflicts:
 228        conflicts_list = ", ".join(f"'{k}'" for k in sorted(conflicts))
 229        raise ValueError(
 230            f"opentelemetry config conflict: {conflicts_list} defined differently in {base_path} and {new_path}"
 231        )
 232
 233    if "metrics" in merged:
 234        merged["metrics"]["attributes"] = merged_attrs
 235    elif merged_attrs:
 236        merged["metrics"] = {"attributes": merged_attrs}
 237
 238    return merged
 239
 240
 241def _merge_configs(
 242    base_config: dict[str, Any],
 243    new_config: dict[str, Any],
 244    base_path: str,
 245    new_path: str,
 246) -> dict[str, Any]:
 247    """
 248    Merge two config dictionaries with field-specific strategies.
 249
 250    Different config fields have different merge strategies:
 251    - profiles: Shallow merge with idempotent check (uses _merge_profiles)
 252    - path_mapping: Flat dict merge with idempotent check
 253    - experimental_features: Flat dict merge with idempotent check
 254    - opentelemetry: Hybrid merge (attributes concatenate, others idempotent)
 255    - cache, posix: Global configs, idempotent if identical, error if different
 256
 257    :param base_config: Base configuration dictionary
 258    :param new_config: New configuration to merge
 259    :param base_path: Path to base config file (for error messages)
 260    :param new_path: Path to new config file (for error messages)
 261    :return: Merged configuration dictionary
 262    :raises ValueError: If conflicts are detected
 263    """
 264    result = {}
 265
 266    all_keys = set(base_config.keys()) | set(new_config.keys())
 267
 268    for key in all_keys:
 269        base_value = base_config.get(key)
 270        new_value = new_config.get(key)
 271
 272        # Key only in base
 273        if key not in new_config:
 274            result[key] = base_value
 275            continue
 276
 277        # Key only in new
 278        if key not in base_config:
 279            result[key] = new_value
 280            continue
 281
 282        # Key in both - need to merge or detect conflict
 283        if key == "profiles":
 284            result["profiles"] = _merge_profiles(base_value or {}, new_value or {}, base_path, new_path)
 285
 286        elif key in ("path_mapping", "experimental_features"):
 287            merged, conflicts = merge_dictionaries_no_overwrite(
 288                (base_value or {}).copy(), new_value or {}, allow_idempotent=True
 289            )
 290            if conflicts:
 291                conflicts_list = ", ".join(f"'{k}'" for k in sorted(conflicts))
 292                raise ValueError(
 293                    f"Config merge conflict: {conflicts_list} have different values in {base_path} and {new_path}"
 294                )
 295            result[key] = merged
 296
 297        elif key == "opentelemetry":
 298            result["opentelemetry"] = _merge_opentelemetry(base_value or {}, new_value or {}, base_path, new_path)
 299
 300        elif key in ("cache", "posix"):
 301            if base_value != new_value:
 302                raise ValueError(f"'{key}' defined differently in {base_path} and {new_path}")
 303            result[key] = base_value
 304
 305        elif key == "include":
 306            # 'include' is processed by _load_and_merge_includes, not part of final config
 307            pass
 308
 309        else:
 310            # This should never happen and all top level fields must have explicit handling above
 311            raise ValueError(f"Unknown field '{key}' in config file.")
 312
 313    return result
 314
 315
 316def _load_and_merge_includes(
 317    main_config_path: str,
 318    main_config_dict: dict[str, Any],
 319) -> dict[str, Any]:
 320    """
 321    Load and merge included config files.
 322
 323    Processes the 'include' directive in the main config, loading and merging all
 324    specified config files. Only supports one level of includes - included files
 325    cannot themselves have 'include' directives.
 326
 327    :param main_config_path: Absolute path to the main config file
 328    :param main_config_dict: Dictionary loaded from the main config file
 329    :return: Merged configuration dictionary (without 'include' field)
 330    :raises ValueError: If include file not found, malformed, or contains nested includes
 331    """
 332    include_paths = main_config_dict.get("include", [])
 333
 334    if not include_paths:
 335        return {k: v for k, v in main_config_dict.items() if k != "include"}
 336
 337    merged_config = {k: v for k, v in main_config_dict.items() if k != "include"}
 338
 339    for include_path in include_paths:
 340        resolved_path = _resolve_include_path(include_path, main_config_path)
 341
 342        if not os.path.exists(resolved_path):
 343            raise ValueError(f"Included config file not found: {resolved_path} (from {main_config_path})")
 344
 345        try:
 346            with open(resolved_path) as f:
 347                if resolved_path.endswith(".json"):
 348                    included_config = json.load(f)
 349                else:
 350                    included_config = yaml.safe_load(f)
 351        except Exception as e:
 352            raise ValueError(f"Failed to load included config {resolved_path}: {e}")
 353
 354        validate_config(included_config)
 355        if "include" in included_config:
 356            raise ValueError(f"Nested includes not allowed: {resolved_path} contains 'include' directive")
 357
 358        merged_config = _merge_configs(merged_config, included_config, main_config_path, resolved_path)
 359
 360    return merged_config
 361
 362
 363def _find_config_file_paths() -> tuple[str, ...]:
 364    """
 365    Get configuration file search paths.
 366
 367    Returns paths in order of precedence:
 368
 369    1. User-specific config (${XDG_CONFIG_HOME}/msc/, ${HOME}/, ${HOME}/.config/msc/)
 370    2. System-wide configs (${XDG_CONFIG_DIRS}/msc/, /etc/xdg, /etc/)
 371    """
 372    paths = []
 373
 374    # 1. User-specific configuration directory
 375    xdg_config_home = os.getenv("XDG_CONFIG_HOME")
 376
 377    if xdg_config_home:
 378        paths.extend(
 379            [
 380                os.path.join(xdg_config_home, "msc", "config.yaml"),
 381                os.path.join(xdg_config_home, "msc", "config.json"),
 382            ]
 383        )
 384
 385    user_home = os.getenv("HOME")
 386
 387    if user_home:
 388        paths.extend(
 389            [
 390                os.path.join(user_home, ".msc_config.yaml"),
 391                os.path.join(user_home, ".msc_config.json"),
 392                os.path.join(user_home, ".config", "msc", "config.yaml"),
 393                os.path.join(user_home, ".config", "msc", "config.json"),
 394            ]
 395        )
 396
 397    # 2. System-wide configuration directories
 398    xdg_config_dirs = os.getenv("XDG_CONFIG_DIRS")
 399    if not xdg_config_dirs:
 400        xdg_config_dirs = "/etc/xdg"
 401
 402    for config_dir in xdg_config_dirs.split(":"):
 403        config_dir = config_dir.strip()
 404        if config_dir:
 405            paths.extend(
 406                [
 407                    os.path.join(config_dir, "msc", "config.yaml"),
 408                    os.path.join(config_dir, "msc", "config.json"),
 409                ]
 410            )
 411
 412    paths.extend(
 413        [
 414            "/etc/msc_config.yaml",
 415            "/etc/msc_config.json",
 416        ]
 417    )
 418
 419    return tuple(paths)
 420
 421
 422PACKAGE_NAME = "multistorageclient"
 423
 424logger = logging.getLogger(__name__)
 425
 426
 427class ImmutableDict(dict):
 428    """
 429    Immutable dictionary that raises an error when attempting to modify it.
 430    """
 431
 432    def __init__(self, *args, **kwargs):
 433        super().__init__(*args, **kwargs)
 434
 435        # Recursively freeze nested structures
 436        for key, value in list(super().items()):
 437            if isinstance(value, dict) and not isinstance(value, ImmutableDict):
 438                super().__setitem__(key, ImmutableDict(value))
 439            elif isinstance(value, list):
 440                super().__setitem__(key, self._freeze_list(value))
 441
 442    @staticmethod
 443    def _freeze_list(lst):
 444        """
 445        Convert list to tuple, freezing nested dicts recursively.
 446        """
 447        frozen = []
 448        for item in lst:
 449            if isinstance(item, dict):
 450                frozen.append(ImmutableDict(item))
 451            elif isinstance(item, list):
 452                frozen.append(ImmutableDict._freeze_list(item))
 453            else:
 454                frozen.append(item)
 455        return tuple(frozen)
 456
 457    def __deepcopy__(self, memo):
 458        """
 459        Return a regular mutable dict when deepcopy is called.
 460        """
 461        # dict(self) would keep the frozen tuples; _copy_value converts nested structures back to dicts and lists.
 462        return copy.deepcopy(self._copy_value(self), memo)
 463
 464    def __reduce__(self):
 465        """
 466        Support for pickle serialization.
 467        """
 468        return (self.__class__, (dict(self),))
 469
 470    def _copy_value(self, value):
 471        """
 472        Convert frozen structures back to mutable equivalents.
 473        """
 474        if isinstance(value, ImmutableDict):
 475            return {k: self._copy_value(v) for k, v in value.items()}
 476        elif isinstance(value, tuple):
 477            # Check if it was originally a list (frozen by _freeze_list)
 478            return [self._copy_value(item) for item in value]
 479        else:
 480            return value
 481
 482    def __getitem__(self, key):
 483        """
 484        Return a mutable copy of the value.
 485        """
 486        value = super().__getitem__(key)
 487        return self._copy_value(value)
 488
 489    def get(self, key, default=None):
 490        """
 491        Return a mutable copy of the value.
 492        """
 493        return self[key] if key in self else default  # noqa: SIM401
 494
 495    def __setitem__(self, key, value):
 496        raise TypeError("ImmutableDict is immutable")
 497
 498    def __delitem__(self, key):
 499        raise TypeError("ImmutableDict is immutable")
 500
 501    def __ior__(self, other):
 502        raise TypeError("ImmutableDict is immutable")
 503
 504    def clear(self):
 505        raise TypeError("ImmutableDict is immutable")
 506
 507    def pop(self, *args):
 508        raise TypeError("ImmutableDict is immutable")
 509
 510    def popitem(self):
 511        raise TypeError("ImmutableDict is immutable")
 512
 513    def setdefault(self, key, default=None):
 514        raise TypeError("ImmutableDict is immutable")
 515
 516    def update(self, *args, **kwargs):
 517        raise TypeError("ImmutableDict is immutable")
 518
 519
 520class SimpleProviderBundle(ProviderBundle):
 521    def __init__(
 522        self,
 523        storage_provider_config: StorageProviderConfig,
 524        credentials_provider: CredentialsProvider | None = None,
 525        metadata_provider: MetadataProvider | None = None,
 526        replicas: list[Replica] | None = None,
 527    ):
 528        if replicas is None:
 529            replicas = []
 530
 531        self._storage_provider_config = storage_provider_config
 532        self._credentials_provider = credentials_provider
 533        self._metadata_provider = metadata_provider
 534        self._replicas = replicas
 535
 536    @property
 537    def storage_provider_config(self) -> StorageProviderConfig:
 538        return self._storage_provider_config
 539
 540    @property
 541    def credentials_provider(self) -> CredentialsProvider | None:
 542        return self._credentials_provider
 543
 544    @property
 545    def metadata_provider(self) -> MetadataProvider | None:
 546        return self._metadata_provider
 547
 548    @property
 549    def replicas(self) -> list[Replica]:
 550        return self._replicas
 551
 552
 553class SimpleProviderBundleV2(ProviderBundleV2):
 554    def __init__(self, storage_backends: dict[str, StorageBackend], metadata_provider: MetadataProvider | None = None):
 555        self._storage_backends = storage_backends
 556        self._metadata_provider = metadata_provider
 557
 558    @staticmethod
 559    def from_v1_bundle(profile_name: str, v1_bundle: ProviderBundle) -> "SimpleProviderBundleV2":
 560        backend = StorageBackend(
 561            storage_provider_config=v1_bundle.storage_provider_config,
 562            credentials_provider=v1_bundle.credentials_provider,
 563            replicas=v1_bundle.replicas,
 564        )
 565
 566        return SimpleProviderBundleV2(
 567            storage_backends={profile_name: backend},
 568            metadata_provider=v1_bundle.metadata_provider,
 569        )
 570
 571    @property
 572    def storage_backends(self) -> dict[str, StorageBackend]:
 573        return self._storage_backends
 574
 575    @property
 576    def metadata_provider(self) -> MetadataProvider | None:
 577        return self._metadata_provider
 578
 579
 580DEFAULT_CACHE_REFRESH_INTERVAL = 300
 581
 582
 583class StorageClientConfigLoader:
 584    _provider_bundle: ProviderBundleV2
 585    _resolved_config_dict: dict[str, Any]
 586    _profiles: dict[str, Any]
 587    _profile: str
 588    _profile_dict: dict[str, Any]
 589    _opentelemetry_dict: dict[str, Any] | None
 590    _telemetry_provider: Callable[[], Telemetry] | None
 591    _cache_dict: dict[str, Any] | None
 592
 593    def __init__(
 594        self,
 595        config_dict: dict[str, Any],
 596        profile: str = RESERVED_POSIX_PROFILE_NAME,
 597        provider_bundle: ProviderBundle | ProviderBundleV2 | None = None,
 598        telemetry_provider: Callable[[], Telemetry] | None = None,
 599    ) -> None:
 600        """
 601        Initializes a :py:class:`StorageClientConfigLoader` to create a
 602        StorageClientConfig. Components are built using the ``config_dict`` and
 603        profile, but a pre-built provider_bundle takes precedence.
 604
 605        :param config_dict: Dictionary of configuration options.
 606        :param profile: Name of profile in ``config_dict`` to use to build configuration.
 607        :param provider_bundle: Optional pre-built :py:class:`multistorageclient.types.ProviderBundle` or :py:class:`multistorageclient.types.ProviderBundleV2`, takes precedence over ``config_dict``.
 608        :param telemetry_provider: A function that provides a telemetry instance. The function must be defined at the top level of a module to work with pickling.
 609        """
 610        # Interpolates all environment variables into actual values.
 611        config_dict = expand_env_vars(config_dict)
 612        self._resolved_config_dict = ImmutableDict(config_dict)
 613
 614        self._profiles = config_dict.get("profiles", {})
 615
 616        if RESERVED_POSIX_PROFILE_NAME not in self._profiles:
 617            # Assign the default POSIX profile
 618            self._profiles[RESERVED_POSIX_PROFILE_NAME] = DEFAULT_POSIX_PROFILE["profiles"][RESERVED_POSIX_PROFILE_NAME]
 619        else:
 620            # Cannot override default POSIX profile
 621            if (
 622                self._profiles[RESERVED_POSIX_PROFILE_NAME]
 623                != DEFAULT_POSIX_PROFILE["profiles"][RESERVED_POSIX_PROFILE_NAME]
 624            ):
 625                raise ValueError(f'Cannot override "{RESERVED_POSIX_PROFILE_NAME}" profile with different settings.')
 626
 627        profile_dict = self._profiles.get(profile)
 628
 629        if not profile_dict:
 630            raise ValueError(f"Profile {profile} not found; available profiles: {list(self._profiles.keys())}")
 631
 632        self._profile = profile
 633        self._profile_dict = ImmutableDict(profile_dict)
 634
 635        self._opentelemetry_dict = config_dict.get("opentelemetry", None)
 636        # Multiprocessing unpickles during the Python interpreter's bootstrap phase for new processes.
 637        # New processes (e.g. multiprocessing manager server) can't be created during this phase.
 638        #
 639        # Pass thunks everywhere instead for lazy telemetry initialization.
 640        self._telemetry_provider = telemetry_provider or telemetry_init
 641
 642        self._cache_dict = config_dict.get("cache", None)
 643
 644        self._provider_bundle = self._build_provider_bundle(provider_bundle)
 645        self._inject_profiles_from_bundle()
 646
 647    def _build_storage_provider(
 648        self,
 649        storage_provider_name: str,
 650        storage_options: dict[str, Any] | None = None,
 651        credentials_provider: CredentialsProvider | None = None,
 652    ) -> StorageProvider:
 653        if storage_options is None:
 654            storage_options = {}
 655        if storage_provider_name not in STORAGE_PROVIDER_MAPPING:
 656            raise ValueError(
 657                f"Storage provider {storage_provider_name} is not supported. "
 658                f"Supported providers are: {list(STORAGE_PROVIDER_MAPPING.keys())}"
 659            )
 660        if credentials_provider:
 661            storage_options["credentials_provider"] = credentials_provider
 662        if self._resolved_config_dict is not None:
 663            # Make a deep copy to drop any external references which may be mutated or cause infinite recursion.
 664            storage_options["config_dict"] = copy.deepcopy(self._resolved_config_dict)
 665        if self._telemetry_provider is not None:
 666            storage_options["telemetry_provider"] = self._telemetry_provider
 667        class_name = STORAGE_PROVIDER_MAPPING[storage_provider_name]
 668        module_name = ".providers"
 669        cls = import_class(class_name, module_name, PACKAGE_NAME)
 670        return cls(**storage_options)
 671
 672    def _build_storage_provider_from_profile(self, storage_provider_profile: str):
 673        storage_profile_dict = self._profiles.get(storage_provider_profile)
 674        if not storage_profile_dict:
 675            raise ValueError(
 676                f"Profile '{storage_provider_profile}' referenced by storage_provider_profile does not exist."
 677            )
 678
 679        # Check if metadata provider is configured for this profile
 680        # NOTE: The storage profile for manifests does not support metadata provider (at the moment).
 681        local_metadata_provider_dict = storage_profile_dict.get("metadata_provider", None)
 682        if local_metadata_provider_dict:
 683            raise ValueError(
 684                f"Profile '{storage_provider_profile}' cannot have a metadata provider when used for manifests"
 685            )
 686
 687        # Initialize CredentialsProvider
 688        local_creds_provider_dict = storage_profile_dict.get("credentials_provider", None)
 689        local_creds_provider = self._build_credentials_provider(credentials_provider_dict=local_creds_provider_dict)
 690
 691        # Initialize StorageProvider
 692        local_storage_provider_dict = storage_profile_dict.get("storage_provider", None)
 693        if local_storage_provider_dict:
 694            local_name = local_storage_provider_dict["type"]
 695            local_storage_options = local_storage_provider_dict.get("options", {})
 696        else:
 697            raise ValueError(f"Missing storage_provider in the config for profile {storage_provider_profile}.")
 698
 699        storage_provider = self._build_storage_provider(local_name, local_storage_options, local_creds_provider)
 700        return storage_provider
 701
 702    def _build_credentials_provider(
 703        self,
 704        credentials_provider_dict: dict[str, Any] | None,
 705        storage_options: dict[str, Any] | None = None,
 706    ) -> CredentialsProvider | None:
 707        """
 708        Initializes the CredentialsProvider based on the provided dictionary.
 709
 710        Args:
 711            credentials_provider_dict: Dictionary containing credentials provider configuration
 712            storage_options: Storage provider options required by some credentials providers to scope the credentials.
 713        """
 714        if not credentials_provider_dict:
 715            return None
 716
 717        if credentials_provider_dict["type"] not in CREDENTIALS_PROVIDER_MAPPING:
 718            # Fully qualified class path case
 719            class_type = credentials_provider_dict["type"]
 720            module_name, class_name = class_type.rsplit(".", 1)
 721            cls = import_class(class_name, module_name)
 722        else:
 723            # Mapped class name case
 724            class_name = CREDENTIALS_PROVIDER_MAPPING[credentials_provider_dict["type"]]
 725            module_name = ".providers"
 726            cls = import_class(class_name, module_name, PACKAGE_NAME)
 727
 728        # Propagate storage provider options to credentials provider since they may be
 729        # required by some credentials providers to scope the credentials.
 730        import inspect
 731
 732        init_params = list(inspect.signature(cls.__init__).parameters)[1:]  # skip 'self'
 733        options = credentials_provider_dict.get("options", {})
 734        if storage_options:
 735            for storage_provider_option in storage_options:
 736                if storage_provider_option in init_params and storage_provider_option not in options:
 737                    options[storage_provider_option] = storage_options[storage_provider_option]
 738
 739        return cls(**options)
 740
 741    def _build_provider_bundle_from_config(self, profile_dict: dict[str, Any]) -> ProviderBundle:
 742        # Initialize StorageProvider
 743        storage_provider_dict = profile_dict.get("storage_provider", None)
 744        if storage_provider_dict:
 745            storage_provider_name = storage_provider_dict["type"]
 746            storage_options = storage_provider_dict.get("options", {})
 747        else:
 748            raise ValueError("Missing storage_provider in the config.")
 749
 750        # Initialize CredentialsProvider
 751        # It is prudent to assume that in some cases, the credentials provider
 752        # will provide credentials scoped to specific base_path.
 753        # So we need to pass the storage_options to the credentials provider.
 754        credentials_provider_dict = profile_dict.get("credentials_provider", None)
 755        credentials_provider = self._build_credentials_provider(
 756            credentials_provider_dict=credentials_provider_dict,
 757            storage_options=storage_options,
 758        )
 759
 760        # Initialize MetadataProvider
 761        metadata_provider_dict = profile_dict.get("metadata_provider", None)
 762        metadata_provider = None
 763        if metadata_provider_dict:
 764            if metadata_provider_dict["type"] == "manifest":
 765                metadata_options = metadata_provider_dict.get("options", {})
 766                # If MetadataProvider has a reference to a different storage provider profile
 767                storage_provider_profile = metadata_options.pop("storage_provider_profile", None)
 768                if storage_provider_profile:
 769                    storage_provider = self._build_storage_provider_from_profile(storage_provider_profile)
 770                else:
 771                    storage_provider = self._build_storage_provider(
 772                        storage_provider_name, storage_options, credentials_provider
 773                    )
 774
 775                metadata_provider = ManifestMetadataProvider(storage_provider, **metadata_options)
 776            else:
 777                class_type = metadata_provider_dict["type"]
 778                if "." not in class_type:
 779                    raise ValueError(
 780                        f"Expected a fully qualified class name (e.g., 'module.ClassName'); got '{class_type}'."
 781                    )
 782                module_name, class_name = class_type.rsplit(".", 1)
 783                cls = import_class(class_name, module_name)
 784                options = metadata_provider_dict.get("options", {})
 785                metadata_provider = cls(**options)
 786
 787        # Build replicas if configured
 788        replicas_config = profile_dict.get("replicas", [])
 789        replicas = []
 790        if replicas_config:
 791            for replica_dict in replicas_config:
 792                replicas.append(
 793                    Replica(
 794                        replica_profile=replica_dict["replica_profile"],
 795                        read_priority=replica_dict["read_priority"],
 796                        read_only=replica_dict.get("read_only", False),
 797                    )
 798                )
 799
 800            # Sort replicas by read_priority
 801            replicas.sort(key=lambda r: r.read_priority)
 802
 803        return SimpleProviderBundle(
 804            storage_provider_config=StorageProviderConfig(storage_provider_name, storage_options),
 805            credentials_provider=credentials_provider,
 806            metadata_provider=metadata_provider,
 807            replicas=replicas,
 808        )
 809
 810    def _build_provider_bundle_from_extension(self, provider_bundle_dict: dict[str, Any]) -> ProviderBundle:
 811        class_type = provider_bundle_dict["type"]
 812        module_name, class_name = class_type.rsplit(".", 1)
 813        cls = import_class(class_name, module_name)
 814        options = provider_bundle_dict.get("options", {})
 815        return cls(**options)
 816
 817    def _build_provider_bundle(self, provider_bundle: ProviderBundle | ProviderBundleV2 | None) -> ProviderBundleV2:
 818        if provider_bundle:
 819            bundle = provider_bundle
 820        else:
 821            provider_bundle_dict = self._profile_dict.get("provider_bundle", None)
 822            if provider_bundle_dict:
 823                bundle = self._build_provider_bundle_from_extension(provider_bundle_dict)
 824            else:
 825                bundle = self._build_provider_bundle_from_config(self._profile_dict)
 826
 827        if isinstance(bundle, ProviderBundle) and not isinstance(bundle, ProviderBundleV2):
 828            bundle = SimpleProviderBundleV2.from_v1_bundle(self._profile, bundle)
 829
 830        return bundle
 831
 832    def _inject_profiles_from_bundle(self) -> None:
 833        """
 834        Inject child profiles and build child configs for multi-backend configurations.
 835
 836        For ProviderBundleV2 with multiple backends, this method:
 837        1. Injects child profiles into config dict (needed for replica initialization)
 838        2. Pre-builds child configs so CompositeStorageClient can use them directly
 839
 840        Profile injection is required because SingleStorageClient._initialize_replicas()
 841        uses StorageClientConfig.from_dict() to create replica clients, which looks up
 842        profiles in the config dict.
 843        """
 844        self._child_configs: dict[str, StorageClientConfig] | None = None
 845
 846        backends = self._provider_bundle.storage_backends
 847        if len(backends) > 1:
 848            # First, inject all child profiles into config dict (needed for replica lookup)
 849            profiles = copy.deepcopy(self._profiles)
 850            for child_name, backend in backends.items():
 851                child_profile_dict = {
 852                    "storage_provider": {
 853                        "type": backend.storage_provider_config.type,
 854                        "options": backend.storage_provider_config.options,
 855                    }
 856                }
 857
 858                if child_name in profiles:
 859                    # Profile already exists - check if it matches what we would inject
 860                    existing = profiles[child_name]
 861                    if existing.get("storage_provider") != child_profile_dict["storage_provider"]:
 862                        raise ValueError(
 863                            f"Profile '{child_name}' already exists in configuration with different settings."
 864                        )
 865                else:
 866                    profiles[child_name] = child_profile_dict
 867
 868            # Update config dict BEFORE building child configs
 869            resolved_config_dict = copy.deepcopy(self._resolved_config_dict)
 870            resolved_config_dict["profiles"] = profiles
 871            self._profiles = ImmutableDict(profiles)
 872            self._resolved_config_dict = ImmutableDict(resolved_config_dict)
 873
 874            # Now build child configs (they will get the updated config dict)
 875            retry_config = self._build_retry_config()
 876            child_configs: dict[str, StorageClientConfig] = {}
 877            for child_name, backend in backends.items():
 878                storage_provider = self._build_storage_provider(
 879                    backend.storage_provider_config.type,
 880                    backend.storage_provider_config.options,
 881                    backend.credentials_provider,
 882                )
 883
 884                child_config = StorageClientConfig(
 885                    profile=child_name,
 886                    storage_provider=storage_provider,
 887                    credentials_provider=backend.credentials_provider,
 888                    storage_provider_profiles=None,
 889                    child_configs=None,
 890                    metadata_provider=None,
 891                    cache_config=None,
 892                    cache_manager=None,
 893                    retry_config=retry_config,
 894                    telemetry_provider=self._telemetry_provider,
 895                    replicas=backend.replicas,
 896                    autocommit_config=None,
 897                )
 898                child_config._config_dict = self._resolved_config_dict
 899                child_configs[child_name] = child_config
 900
 901            self._child_configs = child_configs
 902
 903    def _build_retry_config(self) -> RetryConfig:
 904        """Build retry config from profile dict."""
 905        retry_config_dict = self._profile_dict.get("retry", None)
 906        if retry_config_dict:
 907            attempts = retry_config_dict.get("attempts", DEFAULT_RETRY_ATTEMPTS)
 908            delay = retry_config_dict.get("delay", DEFAULT_RETRY_DELAY)
 909            backoff_multiplier = retry_config_dict.get("backoff_multiplier", DEFAULT_RETRY_BACKOFF_MULTIPLIER)
 910            return RetryConfig(attempts=attempts, delay=delay, backoff_multiplier=backoff_multiplier)
 911        else:
 912            return RetryConfig(
 913                attempts=DEFAULT_RETRY_ATTEMPTS,
 914                delay=DEFAULT_RETRY_DELAY,
 915                backoff_multiplier=DEFAULT_RETRY_BACKOFF_MULTIPLIER,
 916            )
 917
 918    def _build_autocommit_config(self) -> AutoCommitConfig:
 919        """Build autocommit config from profile dict."""
 920        autocommit_dict = self._profile_dict.get("autocommit", None)
 921        if autocommit_dict:
 922            interval_minutes = autocommit_dict.get("interval_minutes", None)
 923            at_exit = autocommit_dict.get("at_exit", False)
 924            return AutoCommitConfig(interval_minutes=interval_minutes, at_exit=at_exit)
 925        return AutoCommitConfig()
 926
 927    def _verify_cache_config(self, cache_dict: dict[str, Any]) -> None:
 928        if "size_mb" in cache_dict:
 929            raise ValueError(
 930                "The 'size_mb' property is no longer supported. \n"
 931                "Please use 'size' with a unit suffix (M, G, T) instead of size_mb.\n"
 932                "Example configuration:\n"
 933                "cache:\n"
 934                "  size: 500G                    # Optional: Maximum cache size (default: 10G)\n"
 935                "  cache_line_size: 64M          # Optional: Chunk size for partial file caching (default: 64M)\n"
 936                "  check_source_version: true    # Optional: Use ETag for cache validation (default: true)\n"
 937                "  location: /tmp/msc_cache      # Optional: Cache directory path (default: system tempdir + '/msc_cache')\n"
 938                "  eviction_policy:               # Optional: Cache eviction policy\n"
 939                "    policy: fifo                 # Optional: Policy type: lru, mru, fifo, random, no_eviction (default: fifo)\n"
 940                "    refresh_interval: 300        # Optional: Cache refresh interval in seconds (default: 300)\n"
 941            )
 942
 943        # Validate that cache_line_size doesn't exceed cache size
 944        cache_size_str = cache_dict.get("size", DEFAULT_CACHE_SIZE)
 945        cache_line_size_str = cache_dict.get("cache_line_size", DEFAULT_CACHE_LINE_SIZE)
 946
 947        # Use CacheConfig to convert sizes to bytes for comparison
 948        temp_cache_config = CacheConfig(
 949            size=cache_size_str,
 950            cache_line_size=cache_line_size_str,
 951        )
 952        cache_size_bytes = temp_cache_config.size_bytes()
 953        cache_line_size_bytes = temp_cache_config.cache_line_size_bytes()
 954
 955        if cache_line_size_bytes > cache_size_bytes:
 956            raise ValueError(
 957                f"cache_line_size ({cache_line_size_str}) exceeds cache size ({cache_size_str}). "
 958                f"cache_line_size must be less than or equal to cache size. "
 959                f"Consider increasing cache size or decreasing cache_line_size."
 960            )
 961
 962    def _validate_replicas(self, replicas: list[Replica]) -> None:
 963        """
 964        Validates that replica profiles do not have their own replicas configuration.
 965
 966        This prevents circular references where a replica profile could reference
 967        another profile that also has replicas, creating an infinite loop.
 968
 969        :param replicas: The list of Replica objects to validate
 970        :raises ValueError: If any replica profile has its own replicas configuration
 971        """
 972        for replica in replicas:
 973            replica_profile_name = replica.replica_profile
 974
 975            # Check that replica profile is not the same as the current profile
 976            if replica_profile_name == self._profile:
 977                raise ValueError(
 978                    f"Replica profile {replica_profile_name} cannot be the same as the profile {self._profile}."
 979                )
 980
 981            # Check if the replica profile exists in the configuration
 982            if replica_profile_name not in self._profiles:
 983                raise ValueError(f"Replica profile '{replica_profile_name}' not found in configuration")
 984
 985            # Get the replica profile configuration
 986            replica_profile_dict = self._profiles[replica_profile_name]
 987
 988            # Check if the replica profile has its own replicas configuration
 989            if replica_profile_dict.get("replicas"):
 990                raise ValueError(
 991                    f"Invalid replica configuration: profile '{replica_profile_name}' has its own replicas. "
 992                    f"This creates a circular reference which is not allowed."
 993                )
 994
 995    def build_config(self) -> "StorageClientConfig":
 996        bundle = self._provider_bundle
 997        backends = bundle.storage_backends
 998
 999        if len(backends) > 1:
1000            if not bundle.metadata_provider:
1001                raise ValueError(
1002                    f"Multi-backend configuration for profile '{self._profile}' requires metadata_provider "
1003                    "for routing between storage locations."
1004                )
1005
1006            child_profile_names = list(backends.keys())
1007            retry_config = self._build_retry_config()
1008            autocommit_config = self._build_autocommit_config()
1009
1010            # config for Composite StorageClient
1011            config = StorageClientConfig(
1012                profile=self._profile,
1013                storage_provider=None,
1014                credentials_provider=None,
1015                storage_provider_profiles=child_profile_names,
1016                child_configs=self._child_configs,
1017                metadata_provider=bundle.metadata_provider,
1018                cache_config=None,
1019                cache_manager=None,
1020                retry_config=retry_config,
1021                telemetry_provider=self._telemetry_provider,
1022                replicas=[],
1023                autocommit_config=autocommit_config,
1024            )
1025
1026            config._config_dict = self._resolved_config_dict
1027            return config
1028
1029        # Single-backend (len == 1)
1030        _, backend = next(iter(backends.items()))
1031        storage_provider = self._build_storage_provider(
1032            backend.storage_provider_config.type,
1033            backend.storage_provider_config.options,
1034            backend.credentials_provider,
1035        )
1036        credentials_provider = backend.credentials_provider
1037        metadata_provider = bundle.metadata_provider
1038        replicas = backend.replicas
1039
1040        # Validate replicas to prevent circular references
1041        if replicas:
1042            self._validate_replicas(replicas)
1043
1044        cache_config: CacheConfig | None = None
1045        cache_manager: CacheManager | None = None
1046
1047        # Check if caching is enabled for this profile
1048        caching_enabled = self._profile_dict.get("caching_enabled", False)
1049
1050        if self._cache_dict is not None and caching_enabled:
1051            tempdir = tempfile.gettempdir()
1052            default_location = os.path.join(tempdir, "msc_cache")
1053            location = self._cache_dict.get("location", default_location)
1054
1055            # Check if cache_backend.cache_path is defined
1056            cache_backend = self._cache_dict.get("cache_backend", {})
1057            cache_backend_path = cache_backend.get("cache_path") if cache_backend else None
1058
1059            # Warn if both location and cache_backend.cache_path are defined
1060            if cache_backend_path and self._cache_dict.get("location") is not None:
1061                logger.warning(
1062                    f"Both 'location' and 'cache_backend.cache_path' are defined in cache config. "
1063                    f"Using 'location' ({location}) and ignoring 'cache_backend.cache_path' ({cache_backend_path})."
1064                )
1065            elif cache_backend_path:
1066                # Use cache_backend.cache_path only if location is not explicitly defined
1067                location = cache_backend_path
1068
1069            check_source_version = self._cache_dict.get("check_source_version", True)
1070
1071            if not Path(location).is_absolute():
1072                raise ValueError(f"Cache location must be an absolute path: {location}")
1073
1074            # Initialize cache_dict with default values
1075            cache_dict = self._cache_dict
1076
1077            # Verify cache config
1078            self._verify_cache_config(cache_dict)
1079
1080            # Initialize eviction policy
1081            if "eviction_policy" in cache_dict:
1082                policy = cache_dict["eviction_policy"]["policy"].lower()
1083                purge_factor = cache_dict["eviction_policy"].get("purge_factor", 0)
1084                eviction_policy = EvictionPolicyConfig(
1085                    policy=policy,
1086                    refresh_interval=cache_dict["eviction_policy"].get(
1087                        "refresh_interval", DEFAULT_CACHE_REFRESH_INTERVAL
1088                    ),
1089                    purge_factor=purge_factor,
1090                )
1091            else:
1092                eviction_policy = EvictionPolicyConfig(policy="fifo", refresh_interval=DEFAULT_CACHE_REFRESH_INTERVAL)
1093
1094            # Create cache config from the standardized format
1095            cache_config = CacheConfig(
1096                size=cache_dict.get("size", DEFAULT_CACHE_SIZE),
1097                location=cache_dict.get("location", location),
1098                check_source_version=check_source_version,
1099                prefetch_file=cache_dict.get("prefetch_file", True),
1100                eviction_policy=eviction_policy,
1101                cache_line_size=cache_dict.get("cache_line_size", DEFAULT_CACHE_LINE_SIZE),
1102            )
1103
1104            cache_manager = CacheManager(profile=self._profile, cache_config=cache_config)
1105        elif self._cache_dict is not None and not caching_enabled:
1106            logger.debug(f"Caching is disabled for profile '{self._profile}'")
1107        elif self._cache_dict is None and caching_enabled:
1108            logger.warning(f"Caching is enabled for profile '{self._profile}' but no cache configuration is provided")
1109
1110        retry_config = self._build_retry_config()
1111        autocommit_config = self._build_autocommit_config()
1112
1113        config = StorageClientConfig(
1114            profile=self._profile,
1115            storage_provider=storage_provider,
1116            credentials_provider=credentials_provider,
1117            storage_provider_profiles=None,
1118            child_configs=None,
1119            metadata_provider=metadata_provider,
1120            cache_config=cache_config,
1121            cache_manager=cache_manager,
1122            retry_config=retry_config,
1123            telemetry_provider=self._telemetry_provider,
1124            replicas=replicas,
1125            autocommit_config=autocommit_config,
1126        )
1127
1128        config._config_dict = self._resolved_config_dict
1129        return config
1130
1131
1132class PathMapping:
1133    """
1134    Class to handle path mappings defined in the MSC configuration.
1135
1136    Path mappings create a nested structure of protocol -> bucket -> [(prefix, profile)]
1137    where entries are sorted by prefix length (longest first) for optimal matching.
1138    Longer paths take precedence when matching.
1139    """
1140
1141    def __init__(self):
1142        """Initialize an empty PathMapping."""
1143        self._mapping = defaultdict(lambda: defaultdict(list))
1144
1145    @classmethod
1146    def from_config(cls, config_dict: dict[str, Any] | None = None) -> "PathMapping":
1147        """
1148        Create a PathMapping instance from configuration dictionary.
1149
1150        :param config_dict: Configuration dictionary, if None the config will be loaded
1151        :return: A PathMapping instance with processed mappings
1152        """
1153        if config_dict is None:
1154            # Import locally to avoid circular imports
1155            from multistorageclient.config import StorageClientConfig
1156
1157            config_dict, _ = StorageClientConfig.read_msc_config()
1158
1159        if not config_dict:
1160            return cls()
1161
1162        instance = cls()
1163        instance._load_mapping(config_dict)
1164        return instance
1165
1166    def _load_mapping(self, config_dict: dict[str, Any]) -> None:
1167        """
1168        Load path mapping from a configuration dictionary.
1169
1170        :param config_dict: Configuration dictionary containing path mapping
1171        """
1172        # Get the path_mapping section
1173        path_mapping = config_dict.get("path_mapping", {})
1174        if path_mapping is None:
1175            return
1176
1177        # Process each mapping
1178        for source_path, dest_path in path_mapping.items():
1179            # Validate format
1180            if not source_path.endswith("/"):
1181                continue
1182            if not dest_path.startswith(MSC_PROTOCOL):
1183                continue
1184            if not dest_path.endswith("/"):
1185                continue
1186
1187            # Extract the destination profile
1188            pr_dest = urlparse(dest_path)
1189            dest_profile = pr_dest.netloc
1190
1191            # Parse the source path
1192            pr = urlparse(source_path)
1193            protocol = pr.scheme.lower() if pr.scheme else "file"
1194
1195            if protocol == "file" or source_path.startswith("/"):
1196                # For file or absolute paths, use the whole path as the prefix
1197                # and leave bucket empty
1198                bucket = ""
1199                prefix = source_path if source_path.startswith("/") else build_full_path(source_path, pr)
1200            else:
1201                # For object storage, extract bucket and prefix
1202                bucket = pr.netloc
1203                prefix = build_full_path(source_path, pr)
1204                prefix = prefix.removeprefix("/")
1205
1206            # Add the mapping to the nested dict
1207            self._mapping[protocol][bucket].append((prefix, dest_profile))
1208
1209        # Sort each bucket's prefixes by length (longest first) for optimal matching
1210        for protocol, buckets in self._mapping.items():
1211            for bucket, prefixes in buckets.items():
1212                self._mapping[protocol][bucket] = sorted(prefixes, key=lambda x: len(x[0]), reverse=True)
1213
1214    def find_mapping(self, url: str) -> tuple[str, str] | None:
1215        """
1216        Find the best matching mapping for the given URL.
1217
1218        :param url: URL to find matching mapping for
1219        :return: Tuple of (profile_name, translated_path) if a match is found, None otherwise
1220        """
1221        # Parse the URL
1222        pr = urlparse(url)
1223        protocol = pr.scheme.lower() if pr.scheme else "file"
1224
1225        # For file paths or absolute paths
1226        if protocol == "file" or url.startswith("/"):
1227            path = url if url.startswith("/") else build_full_path(url, pr)
1228
1229            possible_mapping = self._mapping[protocol][""] if protocol in self._mapping else []
1230
1231            # Check each prefix (already sorted by length, longest first)
1232            for prefix, profile in possible_mapping:
1233                if path.startswith(prefix):
1234                    # Calculate the relative path
1235                    rel_path = path[len(prefix) :]
1236                    if not rel_path.startswith("/"):
1237                        rel_path = "/" + rel_path
1238                    return profile, rel_path
1239
1240            return None
1241
1242        # For object storage
1243        bucket = pr.netloc
1244        path = build_full_path(url, pr)
1245        path = path.removeprefix("/")
1246
1247        # Check bucket-specific mapping
1248        possible_mapping = (
1249            self._mapping[protocol][bucket] if (protocol in self._mapping and bucket in self._mapping[protocol]) else []
1250        )
1251
1252        # Check each prefix (already sorted by length, longest first)
1253        for prefix, profile in possible_mapping:
1254            # matching prefix
1255            if path.startswith(prefix):
1256                rel_path = path[len(prefix) :]
1257                # Remove leading slash if present
1258                rel_path = rel_path.removeprefix("/")
1259
1260                return profile, rel_path
1261
1262        return None
1263
1264
[docs] 1265class StorageClientConfig: 1266 """ 1267 Configuration class for the :py:class:`multistorageclient.StorageClient`. 1268 """ 1269 1270 profile: str 1271 storage_provider: StorageProvider | None 1272 credentials_provider: CredentialsProvider | None 1273 storage_provider_profiles: list[str] | None 1274 child_configs: dict[str, "StorageClientConfig"] | None 1275 metadata_provider: MetadataProvider | None 1276 cache_config: CacheConfig | None 1277 cache_manager: CacheManager | None 1278 retry_config: RetryConfig | None 1279 telemetry_provider: Callable[[], Telemetry] | None 1280 replicas: list[Replica] 1281 autocommit_config: AutoCommitConfig | None 1282 1283 _config_dict: dict[str, Any] | None 1284 1285 def __init__( 1286 self, 1287 profile: str, 1288 storage_provider: StorageProvider | None = None, 1289 credentials_provider: CredentialsProvider | None = None, 1290 storage_provider_profiles: list[str] | None = None, 1291 child_configs: dict[str, "StorageClientConfig"] | None = None, 1292 metadata_provider: MetadataProvider | None = None, 1293 cache_config: CacheConfig | None = None, 1294 cache_manager: CacheManager | None = None, 1295 retry_config: RetryConfig | None = None, 1296 telemetry_provider: Callable[[], Telemetry] | None = None, 1297 replicas: list[Replica] | None = None, 1298 autocommit_config: AutoCommitConfig | None = None, 1299 ): 1300 # exactly one of storage_provider or storage_provider_profiles must be set 1301 if storage_provider and storage_provider_profiles: 1302 raise ValueError( 1303 "Cannot specify both storage_provider and storage_provider_profiles. " 1304 "Use storage_provider for SingleStorageClient or storage_provider_profiles for CompositeStorageClient." 1305 ) 1306 if not storage_provider and not storage_provider_profiles: 1307 raise ValueError("Must specify either storage_provider or storage_provider_profiles.") 1308 1309 if replicas is None: 1310 replicas = [] 1311 self.profile = profile 1312 self.storage_provider = storage_provider 1313 self.credentials_provider = credentials_provider 1314 self.storage_provider_profiles = storage_provider_profiles 1315 self.child_configs = child_configs 1316 self.metadata_provider = metadata_provider 1317 self.cache_config = cache_config 1318 self.cache_manager = cache_manager 1319 self.retry_config = retry_config 1320 self.telemetry_provider = telemetry_provider 1321 self.replicas = replicas 1322 self.autocommit_config = autocommit_config 1323
[docs] 1324 @staticmethod 1325 def from_json( 1326 config_json: str, 1327 profile: str = RESERVED_POSIX_PROFILE_NAME, 1328 telemetry_provider: Callable[[], Telemetry] | None = None, 1329 ) -> "StorageClientConfig": 1330 """ 1331 Load a storage client configuration from a JSON string. 1332 1333 :param config_json: Configuration JSON string. 1334 :param profile: Profile to use. 1335 :param telemetry_provider: A function that provides a telemetry instance. The function must be defined at the top level of a module to work with pickling. 1336 """ 1337 config_dict = json.loads(config_json) 1338 return StorageClientConfig.from_dict( 1339 config_dict=config_dict, profile=profile, telemetry_provider=telemetry_provider 1340 )
1341
[docs] 1342 @staticmethod 1343 def from_yaml( 1344 config_yaml: str, 1345 profile: str = RESERVED_POSIX_PROFILE_NAME, 1346 telemetry_provider: Callable[[], Telemetry] | None = None, 1347 ) -> "StorageClientConfig": 1348 """ 1349 Load a storage client configuration from a YAML string. 1350 1351 :param config_yaml: Configuration YAML string. 1352 :param profile: Profile to use. 1353 :param telemetry_provider: A function that provides a telemetry instance. The function must be defined at the top level of a module to work with pickling. 1354 """ 1355 config_dict = yaml.safe_load(config_yaml) or {} 1356 return StorageClientConfig.from_dict( 1357 config_dict=config_dict, profile=profile, telemetry_provider=telemetry_provider 1358 )
1359
[docs] 1360 @staticmethod 1361 def from_dict( 1362 config_dict: dict[str, Any], 1363 profile: str = RESERVED_POSIX_PROFILE_NAME, 1364 skip_validation: bool = False, 1365 telemetry_provider: Callable[[], Telemetry] | None = None, 1366 ) -> "StorageClientConfig": 1367 """ 1368 Load a storage client configuration from a Python dictionary. 1369 1370 :param config_dict: Configuration Python dictionary. 1371 :param profile: Profile to use. 1372 :param skip_validation: Skip configuration schema validation. 1373 :param telemetry_provider: A function that provides a telemetry instance. The function must be defined at the top level of a module to work with pickling. 1374 """ 1375 # Validate the config file with predefined JSON schema 1376 if not skip_validation: 1377 validate_config(config_dict) 1378 1379 # Load config 1380 loader = StorageClientConfigLoader( 1381 config_dict=config_dict, 1382 profile=profile, 1383 telemetry_provider=telemetry_provider, 1384 ) 1385 config = loader.build_config() 1386 1387 return config
1388
[docs] 1389 @staticmethod 1390 def from_file( 1391 config_file_paths: Iterable[str] | None = None, 1392 profile: str = RESERVED_POSIX_PROFILE_NAME, 1393 telemetry_provider: Callable[[], Telemetry] | None = None, 1394 ) -> "StorageClientConfig": 1395 """ 1396 Load a storage client configuration from the first file found. 1397 1398 :param config_file_paths: Configuration file search paths. If omitted, the default search paths are used (see :py:meth:`StorageClientConfig.read_msc_config`). 1399 :param profile: Profile to use. 1400 :param telemetry_provider: A function that provides a telemetry instance. The function must be defined at the top level of a module to work with pickling. 1401 """ 1402 msc_config_dict, msc_config_file = StorageClientConfig.read_msc_config(config_file_paths=config_file_paths) 1403 # Parse rclone config file. 1404 rclone_config_dict, rclone_config_file = read_rclone_config() 1405 1406 # Merge config files. 1407 merged_config, conflicted_keys = merge_dictionaries_no_overwrite(msc_config_dict, rclone_config_dict) 1408 if conflicted_keys: 1409 raise ValueError( 1410 f'Conflicting keys found in configuration files "{msc_config_file}" and "{rclone_config_file}: {conflicted_keys}' 1411 ) 1412 merged_profiles = merged_config.get("profiles", {}) 1413 1414 # Check if profile is in merged_profiles 1415 if profile in merged_profiles: 1416 return StorageClientConfig.from_dict( 1417 config_dict=merged_config, profile=profile, telemetry_provider=telemetry_provider 1418 ) 1419 else: 1420 # Check if profile is the default POSIX profile or an implicit profile 1421 if profile == RESERVED_POSIX_PROFILE_NAME: 1422 implicit_profile_config = DEFAULT_POSIX_PROFILE 1423 elif profile.startswith("_"): 1424 # Handle implicit profiles 1425 parts = profile[1:].split("-", 1) 1426 if len(parts) == 2: 1427 protocol, bucket = parts 1428 # Verify it's a supported protocol 1429 if protocol not in SUPPORTED_IMPLICIT_PROFILE_PROTOCOLS: 1430 raise ValueError(f'Unsupported protocol in implicit profile: "{protocol}"') 1431 implicit_profile_config = create_implicit_profile_config( 1432 profile_name=profile, protocol=protocol, base_path=bucket 1433 ) 1434 else: 1435 raise ValueError(f'Invalid implicit profile format: "{profile}"') 1436 else: 1437 raise ValueError( 1438 f'Profile "{profile}" not found in configuration files. Configuration was checked in ' 1439 f"{msc_config_file or 'MSC config (not found)'} and {rclone_config_file or 'Rclone config (not found)'}. " 1440 f"Please verify that the profile exists and that configuration files are correctly located." 1441 ) 1442 # merge the implicit profile config into the merged config so the cache & observability config can be inherited 1443 if "profiles" not in merged_config: 1444 merged_config["profiles"] = implicit_profile_config["profiles"] 1445 else: 1446 merged_config["profiles"][profile] = implicit_profile_config["profiles"][profile] 1447 # the config is already validated while reading, skip the validation for implicit profiles which start profile with "_" 1448 return StorageClientConfig.from_dict( 1449 config_dict=merged_config, profile=profile, skip_validation=True, telemetry_provider=telemetry_provider 1450 )
1451 1452 @staticmethod 1453 def from_provider_bundle( 1454 config_dict: dict[str, Any], 1455 provider_bundle: ProviderBundle | ProviderBundleV2, 1456 telemetry_provider: Callable[[], Telemetry] | None = None, 1457 ) -> "StorageClientConfig": 1458 loader = StorageClientConfigLoader( 1459 config_dict=config_dict, provider_bundle=provider_bundle, telemetry_provider=telemetry_provider 1460 ) 1461 config = loader.build_config() 1462 config._config_dict = None # Explicitly mark as None to avoid confusing pickling errors 1463 return config 1464
[docs] 1465 @staticmethod 1466 def read_msc_config( 1467 config_file_paths: Iterable[str] | None = None, 1468 ) -> tuple[dict[str, Any] | None, str | None]: 1469 """Get the MSC configuration dictionary and the path of the first file found. 1470 1471 If no config paths are specified, configs are searched in the following order: 1472 1473 1. ``MSC_CONFIG`` environment variable (highest precedence) 1474 2. Standard search paths (user-specified config and system-wide config) 1475 1476 :param config_file_paths: Configuration file search paths. If omitted, the default search paths are used. 1477 :return: Tuple of ``(config_dict, config_file_path)``. ``config_dict`` is the MSC configuration 1478 dictionary or empty dict if no config was found. ``config_file_path`` is the absolute 1479 path of the config file used, or ``None`` if no config file was found. 1480 """ 1481 config_dict: dict[str, Any] = {} 1482 config_file_path: str | None = None 1483 1484 config_file_paths = list(config_file_paths or []) 1485 1486 # Add default paths if none provided. 1487 if len(config_file_paths) == 0: 1488 # Environment variable. 1489 msc_config_env = os.getenv("MSC_CONFIG", None) 1490 if msc_config_env is not None: 1491 config_file_paths.append(msc_config_env) 1492 1493 # Standard search paths. 1494 config_file_paths.extend(_find_config_file_paths()) 1495 1496 # Normalize + absolutize paths. 1497 config_file_paths = [os.path.abspath(path) for path in config_file_paths] 1498 1499 # Log plan. 1500 logger.debug(f"Searching MSC config file paths: {config_file_paths}") 1501 1502 # Load config. 1503 for path in config_file_paths: 1504 if os.path.exists(path): 1505 try: 1506 with open(path) as f: 1507 if path.endswith(".json"): 1508 config_dict = json.load(f) 1509 else: 1510 config_dict = yaml.safe_load(f) 1511 config_file_path = path 1512 # Use the first config file. 1513 break 1514 except Exception as e: 1515 raise ValueError(f"malformed MSC config file: {path}, exception: {e}") 1516 1517 # Log result. 1518 if config_file_path is None: 1519 logger.debug("No MSC config files found in any of the search locations.") 1520 else: 1521 logger.debug(f"Using MSC config file: {config_file_path}") 1522 1523 if config_dict: 1524 validate_config(config_dict) 1525 1526 if "include" in config_dict and config_file_path: 1527 config_dict = _load_and_merge_includes(config_file_path, config_dict) 1528 1529 return config_dict, config_file_path
1530
[docs] 1531 @staticmethod 1532 def read_path_mapping() -> PathMapping: 1533 """ 1534 Get the path mapping defined in the MSC configuration. 1535 1536 Path mappings create a nested structure of protocol -> bucket -> [(prefix, profile)] 1537 where entries are sorted by prefix length (longest first) for optimal matching. 1538 Longer paths take precedence when matching. 1539 1540 :return: A PathMapping instance with translation mappings 1541 """ 1542 try: 1543 return PathMapping.from_config() 1544 except Exception: 1545 # Log the error but continue - this shouldn't stop the application from working 1546 logger.error("Failed to load path_mapping from MSC config") 1547 return PathMapping()
1548 1549 def __getstate__(self) -> dict[str, Any]: 1550 state = self.__dict__.copy() 1551 if not state.get("_config_dict"): 1552 raise ValueError("StorageClientConfig is not serializable") 1553 del state["credentials_provider"] 1554 del state["storage_provider"] 1555 del state["metadata_provider"] 1556 del state["cache_manager"] 1557 del state["replicas"] 1558 del state["child_configs"] 1559 return state 1560 1561 def __setstate__(self, state: dict[str, Any]) -> None: 1562 # Presence checked by __getstate__. 1563 config_dict = state["_config_dict"] 1564 loader = StorageClientConfigLoader( 1565 config_dict=config_dict, 1566 profile=state["profile"], 1567 telemetry_provider=state["telemetry_provider"], 1568 ) 1569 new_config = loader.build_config() 1570 self.profile = new_config.profile 1571 self.storage_provider = new_config.storage_provider 1572 self.credentials_provider = new_config.credentials_provider 1573 self.storage_provider_profiles = new_config.storage_provider_profiles 1574 self.child_configs = new_config.child_configs 1575 self.metadata_provider = new_config.metadata_provider 1576 self.cache_config = new_config.cache_config 1577 self.cache_manager = new_config.cache_manager 1578 self.retry_config = new_config.retry_config 1579 self.telemetry_provider = new_config.telemetry_provider 1580 self._config_dict = config_dict 1581 self.replicas = new_config.replicas 1582 self.autocommit_config = new_config.autocommit_config