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 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 clear(self):
 502        raise TypeError("ImmutableDict is immutable")
 503
 504    def pop(self, *args):
 505        raise TypeError("ImmutableDict is immutable")
 506
 507    def popitem(self):
 508        raise TypeError("ImmutableDict is immutable")
 509
 510    def setdefault(self, key, default=None):
 511        raise TypeError("ImmutableDict is immutable")
 512
 513    def update(self, *args, **kwargs):
 514        raise TypeError("ImmutableDict is immutable")
 515
 516
 517class SimpleProviderBundle(ProviderBundle):
 518    def __init__(
 519        self,
 520        storage_provider_config: StorageProviderConfig,
 521        credentials_provider: CredentialsProvider | None = None,
 522        metadata_provider: MetadataProvider | None = None,
 523        replicas: list[Replica] | None = None,
 524    ):
 525        if replicas is None:
 526            replicas = []
 527
 528        self._storage_provider_config = storage_provider_config
 529        self._credentials_provider = credentials_provider
 530        self._metadata_provider = metadata_provider
 531        self._replicas = replicas
 532
 533    @property
 534    def storage_provider_config(self) -> StorageProviderConfig:
 535        return self._storage_provider_config
 536
 537    @property
 538    def credentials_provider(self) -> CredentialsProvider | None:
 539        return self._credentials_provider
 540
 541    @property
 542    def metadata_provider(self) -> MetadataProvider | None:
 543        return self._metadata_provider
 544
 545    @property
 546    def replicas(self) -> list[Replica]:
 547        return self._replicas
 548
 549
 550class SimpleProviderBundleV2(ProviderBundleV2):
 551    def __init__(self, storage_backends: dict[str, StorageBackend], metadata_provider: MetadataProvider | None = None):
 552        self._storage_backends = storage_backends
 553        self._metadata_provider = metadata_provider
 554
 555    @staticmethod
 556    def from_v1_bundle(profile_name: str, v1_bundle: ProviderBundle) -> "SimpleProviderBundleV2":
 557        backend = StorageBackend(
 558            storage_provider_config=v1_bundle.storage_provider_config,
 559            credentials_provider=v1_bundle.credentials_provider,
 560            replicas=v1_bundle.replicas,
 561        )
 562
 563        return SimpleProviderBundleV2(
 564            storage_backends={profile_name: backend},
 565            metadata_provider=v1_bundle.metadata_provider,
 566        )
 567
 568    @property
 569    def storage_backends(self) -> dict[str, StorageBackend]:
 570        return self._storage_backends
 571
 572    @property
 573    def metadata_provider(self) -> MetadataProvider | None:
 574        return self._metadata_provider
 575
 576
 577DEFAULT_CACHE_REFRESH_INTERVAL = 300
 578
 579
 580class StorageClientConfigLoader:
 581    _provider_bundle: ProviderBundleV2
 582    _resolved_config_dict: dict[str, Any]
 583    _profiles: dict[str, Any]
 584    _profile: str
 585    _profile_dict: dict[str, Any]
 586    _opentelemetry_dict: dict[str, Any] | None
 587    _telemetry_provider: Callable[[], Telemetry] | None
 588    _cache_dict: dict[str, Any] | None
 589
 590    def __init__(
 591        self,
 592        config_dict: dict[str, Any],
 593        profile: str = RESERVED_POSIX_PROFILE_NAME,
 594        provider_bundle: ProviderBundle | ProviderBundleV2 | None = None,
 595        telemetry_provider: Callable[[], Telemetry] | None = None,
 596    ) -> None:
 597        """
 598        Initializes a :py:class:`StorageClientConfigLoader` to create a
 599        StorageClientConfig. Components are built using the ``config_dict`` and
 600        profile, but a pre-built provider_bundle takes precedence.
 601
 602        :param config_dict: Dictionary of configuration options.
 603        :param profile: Name of profile in ``config_dict`` to use to build configuration.
 604        :param provider_bundle: Optional pre-built :py:class:`multistorageclient.types.ProviderBundle` or :py:class:`multistorageclient.types.ProviderBundleV2`, takes precedence over ``config_dict``.
 605        :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.
 606        """
 607        # Interpolates all environment variables into actual values.
 608        config_dict = expand_env_vars(config_dict)
 609        self._resolved_config_dict = ImmutableDict(config_dict)
 610
 611        self._profiles = config_dict.get("profiles", {})
 612
 613        if RESERVED_POSIX_PROFILE_NAME not in self._profiles:
 614            # Assign the default POSIX profile
 615            self._profiles[RESERVED_POSIX_PROFILE_NAME] = DEFAULT_POSIX_PROFILE["profiles"][RESERVED_POSIX_PROFILE_NAME]
 616        else:
 617            # Cannot override default POSIX profile
 618            if (
 619                self._profiles[RESERVED_POSIX_PROFILE_NAME]
 620                != DEFAULT_POSIX_PROFILE["profiles"][RESERVED_POSIX_PROFILE_NAME]
 621            ):
 622                raise ValueError(f'Cannot override "{RESERVED_POSIX_PROFILE_NAME}" profile with different settings.')
 623
 624        profile_dict = self._profiles.get(profile)
 625
 626        if not profile_dict:
 627            raise ValueError(f"Profile {profile} not found; available profiles: {list(self._profiles.keys())}")
 628
 629        self._profile = profile
 630        self._profile_dict = ImmutableDict(profile_dict)
 631
 632        self._opentelemetry_dict = config_dict.get("opentelemetry", None)
 633        # Multiprocessing unpickles during the Python interpreter's bootstrap phase for new processes.
 634        # New processes (e.g. multiprocessing manager server) can't be created during this phase.
 635        #
 636        # Pass thunks everywhere instead for lazy telemetry initialization.
 637        self._telemetry_provider = telemetry_provider or telemetry_init
 638
 639        self._cache_dict = config_dict.get("cache", None)
 640
 641        self._provider_bundle = self._build_provider_bundle(provider_bundle)
 642        self._inject_profiles_from_bundle()
 643
 644    def _build_storage_provider(
 645        self,
 646        storage_provider_name: str,
 647        storage_options: dict[str, Any] | None = None,
 648        credentials_provider: CredentialsProvider | None = None,
 649    ) -> StorageProvider:
 650        if storage_options is None:
 651            storage_options = {}
 652        if storage_provider_name not in STORAGE_PROVIDER_MAPPING:
 653            raise ValueError(
 654                f"Storage provider {storage_provider_name} is not supported. "
 655                f"Supported providers are: {list(STORAGE_PROVIDER_MAPPING.keys())}"
 656            )
 657        if credentials_provider:
 658            storage_options["credentials_provider"] = credentials_provider
 659        if self._resolved_config_dict is not None:
 660            # Make a deep copy to drop any external references which may be mutated or cause infinite recursion.
 661            storage_options["config_dict"] = copy.deepcopy(self._resolved_config_dict)
 662        if self._telemetry_provider is not None:
 663            storage_options["telemetry_provider"] = self._telemetry_provider
 664        class_name = STORAGE_PROVIDER_MAPPING[storage_provider_name]
 665        module_name = ".providers"
 666        cls = import_class(class_name, module_name, PACKAGE_NAME)
 667        return cls(**storage_options)
 668
 669    def _build_storage_provider_from_profile(self, storage_provider_profile: str):
 670        storage_profile_dict = self._profiles.get(storage_provider_profile)
 671        if not storage_profile_dict:
 672            raise ValueError(
 673                f"Profile '{storage_provider_profile}' referenced by storage_provider_profile does not exist."
 674            )
 675
 676        # Check if metadata provider is configured for this profile
 677        # NOTE: The storage profile for manifests does not support metadata provider (at the moment).
 678        local_metadata_provider_dict = storage_profile_dict.get("metadata_provider", None)
 679        if local_metadata_provider_dict:
 680            raise ValueError(
 681                f"Profile '{storage_provider_profile}' cannot have a metadata provider when used for manifests"
 682            )
 683
 684        # Initialize CredentialsProvider
 685        local_creds_provider_dict = storage_profile_dict.get("credentials_provider", None)
 686        local_creds_provider = self._build_credentials_provider(credentials_provider_dict=local_creds_provider_dict)
 687
 688        # Initialize StorageProvider
 689        local_storage_provider_dict = storage_profile_dict.get("storage_provider", None)
 690        if local_storage_provider_dict:
 691            local_name = local_storage_provider_dict["type"]
 692            local_storage_options = local_storage_provider_dict.get("options", {})
 693        else:
 694            raise ValueError(f"Missing storage_provider in the config for profile {storage_provider_profile}.")
 695
 696        storage_provider = self._build_storage_provider(local_name, local_storage_options, local_creds_provider)
 697        return storage_provider
 698
 699    def _build_credentials_provider(
 700        self,
 701        credentials_provider_dict: dict[str, Any] | None,
 702        storage_options: dict[str, Any] | None = None,
 703    ) -> CredentialsProvider | None:
 704        """
 705        Initializes the CredentialsProvider based on the provided dictionary.
 706
 707        Args:
 708            credentials_provider_dict: Dictionary containing credentials provider configuration
 709            storage_options: Storage provider options required by some credentials providers to scope the credentials.
 710        """
 711        if not credentials_provider_dict:
 712            return None
 713
 714        if credentials_provider_dict["type"] not in CREDENTIALS_PROVIDER_MAPPING:
 715            # Fully qualified class path case
 716            class_type = credentials_provider_dict["type"]
 717            module_name, class_name = class_type.rsplit(".", 1)
 718            cls = import_class(class_name, module_name)
 719        else:
 720            # Mapped class name case
 721            class_name = CREDENTIALS_PROVIDER_MAPPING[credentials_provider_dict["type"]]
 722            module_name = ".providers"
 723            cls = import_class(class_name, module_name, PACKAGE_NAME)
 724
 725        # Propagate storage provider options to credentials provider since they may be
 726        # required by some credentials providers to scope the credentials.
 727        import inspect
 728
 729        init_params = list(inspect.signature(cls.__init__).parameters)[1:]  # skip 'self'
 730        options = credentials_provider_dict.get("options", {})
 731        if storage_options:
 732            for storage_provider_option in storage_options:
 733                if storage_provider_option in init_params and storage_provider_option not in options:
 734                    options[storage_provider_option] = storage_options[storage_provider_option]
 735
 736        return cls(**options)
 737
 738    def _build_provider_bundle_from_config(self, profile_dict: dict[str, Any]) -> ProviderBundle:
 739        # Initialize StorageProvider
 740        storage_provider_dict = profile_dict.get("storage_provider", None)
 741        if storage_provider_dict:
 742            storage_provider_name = storage_provider_dict["type"]
 743            storage_options = storage_provider_dict.get("options", {})
 744        else:
 745            raise ValueError("Missing storage_provider in the config.")
 746
 747        # Initialize CredentialsProvider
 748        # It is prudent to assume that in some cases, the credentials provider
 749        # will provide credentials scoped to specific base_path.
 750        # So we need to pass the storage_options to the credentials provider.
 751        credentials_provider_dict = profile_dict.get("credentials_provider", None)
 752        credentials_provider = self._build_credentials_provider(
 753            credentials_provider_dict=credentials_provider_dict,
 754            storage_options=storage_options,
 755        )
 756
 757        # Initialize MetadataProvider
 758        metadata_provider_dict = profile_dict.get("metadata_provider", None)
 759        metadata_provider = None
 760        if metadata_provider_dict:
 761            if metadata_provider_dict["type"] == "manifest":
 762                metadata_options = metadata_provider_dict.get("options", {})
 763                # If MetadataProvider has a reference to a different storage provider profile
 764                storage_provider_profile = metadata_options.pop("storage_provider_profile", None)
 765                if storage_provider_profile:
 766                    storage_provider = self._build_storage_provider_from_profile(storage_provider_profile)
 767                else:
 768                    storage_provider = self._build_storage_provider(
 769                        storage_provider_name, storage_options, credentials_provider
 770                    )
 771
 772                metadata_provider = ManifestMetadataProvider(storage_provider, **metadata_options)
 773            else:
 774                class_type = metadata_provider_dict["type"]
 775                if "." not in class_type:
 776                    raise ValueError(
 777                        f"Expected a fully qualified class name (e.g., 'module.ClassName'); got '{class_type}'."
 778                    )
 779                module_name, class_name = class_type.rsplit(".", 1)
 780                cls = import_class(class_name, module_name)
 781                options = metadata_provider_dict.get("options", {})
 782                metadata_provider = cls(**options)
 783
 784        # Build replicas if configured
 785        replicas_config = profile_dict.get("replicas", [])
 786        replicas = []
 787        if replicas_config:
 788            for replica_dict in replicas_config:
 789                replicas.append(
 790                    Replica(
 791                        replica_profile=replica_dict["replica_profile"],
 792                        read_priority=replica_dict["read_priority"],
 793                        read_only=replica_dict.get("read_only", False),
 794                    )
 795                )
 796
 797            # Sort replicas by read_priority
 798            replicas.sort(key=lambda r: r.read_priority)
 799
 800        return SimpleProviderBundle(
 801            storage_provider_config=StorageProviderConfig(storage_provider_name, storage_options),
 802            credentials_provider=credentials_provider,
 803            metadata_provider=metadata_provider,
 804            replicas=replicas,
 805        )
 806
 807    def _build_provider_bundle_from_extension(self, provider_bundle_dict: dict[str, Any]) -> ProviderBundle:
 808        class_type = provider_bundle_dict["type"]
 809        module_name, class_name = class_type.rsplit(".", 1)
 810        cls = import_class(class_name, module_name)
 811        options = provider_bundle_dict.get("options", {})
 812        return cls(**options)
 813
 814    def _build_provider_bundle(self, provider_bundle: ProviderBundle | ProviderBundleV2 | None) -> ProviderBundleV2:
 815        if provider_bundle:
 816            bundle = provider_bundle
 817        else:
 818            provider_bundle_dict = self._profile_dict.get("provider_bundle", None)
 819            if provider_bundle_dict:
 820                bundle = self._build_provider_bundle_from_extension(provider_bundle_dict)
 821            else:
 822                bundle = self._build_provider_bundle_from_config(self._profile_dict)
 823
 824        if isinstance(bundle, ProviderBundle) and not isinstance(bundle, ProviderBundleV2):
 825            bundle = SimpleProviderBundleV2.from_v1_bundle(self._profile, bundle)
 826
 827        return bundle
 828
 829    def _inject_profiles_from_bundle(self) -> None:
 830        """
 831        Inject child profiles and build child configs for multi-backend configurations.
 832
 833        For ProviderBundleV2 with multiple backends, this method:
 834        1. Injects child profiles into config dict (needed for replica initialization)
 835        2. Pre-builds child configs so CompositeStorageClient can use them directly
 836
 837        Profile injection is required because SingleStorageClient._initialize_replicas()
 838        uses StorageClientConfig.from_dict() to create replica clients, which looks up
 839        profiles in the config dict.
 840        """
 841        self._child_configs: dict[str, StorageClientConfig] | None = None
 842
 843        backends = self._provider_bundle.storage_backends
 844        if len(backends) > 1:
 845            # First, inject all child profiles into config dict (needed for replica lookup)
 846            profiles = copy.deepcopy(self._profiles)
 847            for child_name, backend in backends.items():
 848                child_profile_dict = {
 849                    "storage_provider": {
 850                        "type": backend.storage_provider_config.type,
 851                        "options": backend.storage_provider_config.options,
 852                    }
 853                }
 854
 855                if child_name in profiles:
 856                    # Profile already exists - check if it matches what we would inject
 857                    existing = profiles[child_name]
 858                    if existing.get("storage_provider") != child_profile_dict["storage_provider"]:
 859                        raise ValueError(
 860                            f"Profile '{child_name}' already exists in configuration with different settings."
 861                        )
 862                else:
 863                    profiles[child_name] = child_profile_dict
 864
 865            # Update config dict BEFORE building child configs
 866            resolved_config_dict = copy.deepcopy(self._resolved_config_dict)
 867            resolved_config_dict["profiles"] = profiles
 868            self._profiles = ImmutableDict(profiles)
 869            self._resolved_config_dict = ImmutableDict(resolved_config_dict)
 870
 871            # Now build child configs (they will get the updated config dict)
 872            retry_config = self._build_retry_config()
 873            child_configs: dict[str, StorageClientConfig] = {}
 874            for child_name, backend in backends.items():
 875                storage_provider = self._build_storage_provider(
 876                    backend.storage_provider_config.type,
 877                    backend.storage_provider_config.options,
 878                    backend.credentials_provider,
 879                )
 880
 881                child_config = StorageClientConfig(
 882                    profile=child_name,
 883                    storage_provider=storage_provider,
 884                    credentials_provider=backend.credentials_provider,
 885                    storage_provider_profiles=None,
 886                    child_configs=None,
 887                    metadata_provider=None,
 888                    cache_config=None,
 889                    cache_manager=None,
 890                    retry_config=retry_config,
 891                    telemetry_provider=self._telemetry_provider,
 892                    replicas=backend.replicas,
 893                    autocommit_config=None,
 894                )
 895                child_config._config_dict = self._resolved_config_dict
 896                child_configs[child_name] = child_config
 897
 898            self._child_configs = child_configs
 899
 900    def _build_retry_config(self) -> RetryConfig:
 901        """Build retry config from profile dict."""
 902        retry_config_dict = self._profile_dict.get("retry", None)
 903        if retry_config_dict:
 904            attempts = retry_config_dict.get("attempts", DEFAULT_RETRY_ATTEMPTS)
 905            delay = retry_config_dict.get("delay", DEFAULT_RETRY_DELAY)
 906            backoff_multiplier = retry_config_dict.get("backoff_multiplier", DEFAULT_RETRY_BACKOFF_MULTIPLIER)
 907            return RetryConfig(attempts=attempts, delay=delay, backoff_multiplier=backoff_multiplier)
 908        else:
 909            return RetryConfig(
 910                attempts=DEFAULT_RETRY_ATTEMPTS,
 911                delay=DEFAULT_RETRY_DELAY,
 912                backoff_multiplier=DEFAULT_RETRY_BACKOFF_MULTIPLIER,
 913            )
 914
 915    def _build_autocommit_config(self) -> AutoCommitConfig:
 916        """Build autocommit config from profile dict."""
 917        autocommit_dict = self._profile_dict.get("autocommit", None)
 918        if autocommit_dict:
 919            interval_minutes = autocommit_dict.get("interval_minutes", None)
 920            at_exit = autocommit_dict.get("at_exit", False)
 921            return AutoCommitConfig(interval_minutes=interval_minutes, at_exit=at_exit)
 922        return AutoCommitConfig()
 923
 924    def _verify_cache_config(self, cache_dict: dict[str, Any]) -> None:
 925        if "size_mb" in cache_dict:
 926            raise ValueError(
 927                "The 'size_mb' property is no longer supported. \n"
 928                "Please use 'size' with a unit suffix (M, G, T) instead of size_mb.\n"
 929                "Example configuration:\n"
 930                "cache:\n"
 931                "  size: 500G                    # Optional: Maximum cache size (default: 10G)\n"
 932                "  cache_line_size: 64M          # Optional: Chunk size for partial file caching (default: 64M)\n"
 933                "  check_source_version: true    # Optional: Use ETag for cache validation (default: true)\n"
 934                "  location: /tmp/msc_cache      # Optional: Cache directory path (default: system tempdir + '/msc_cache')\n"
 935                "  eviction_policy:               # Optional: Cache eviction policy\n"
 936                "    policy: fifo                 # Optional: Policy type: lru, mru, fifo, random, no_eviction (default: fifo)\n"
 937                "    refresh_interval: 300        # Optional: Cache refresh interval in seconds (default: 300)\n"
 938            )
 939
 940        # Validate that cache_line_size doesn't exceed cache size
 941        cache_size_str = cache_dict.get("size", DEFAULT_CACHE_SIZE)
 942        cache_line_size_str = cache_dict.get("cache_line_size", DEFAULT_CACHE_LINE_SIZE)
 943
 944        # Use CacheConfig to convert sizes to bytes for comparison
 945        temp_cache_config = CacheConfig(
 946            size=cache_size_str,
 947            cache_line_size=cache_line_size_str,
 948        )
 949        cache_size_bytes = temp_cache_config.size_bytes()
 950        cache_line_size_bytes = temp_cache_config.cache_line_size_bytes()
 951
 952        if cache_line_size_bytes > cache_size_bytes:
 953            raise ValueError(
 954                f"cache_line_size ({cache_line_size_str}) exceeds cache size ({cache_size_str}). "
 955                f"cache_line_size must be less than or equal to cache size. "
 956                f"Consider increasing cache size or decreasing cache_line_size."
 957            )
 958
 959    def _validate_replicas(self, replicas: list[Replica]) -> None:
 960        """
 961        Validates that replica profiles do not have their own replicas configuration.
 962
 963        This prevents circular references where a replica profile could reference
 964        another profile that also has replicas, creating an infinite loop.
 965
 966        :param replicas: The list of Replica objects to validate
 967        :raises ValueError: If any replica profile has its own replicas configuration
 968        """
 969        for replica in replicas:
 970            replica_profile_name = replica.replica_profile
 971
 972            # Check that replica profile is not the same as the current profile
 973            if replica_profile_name == self._profile:
 974                raise ValueError(
 975                    f"Replica profile {replica_profile_name} cannot be the same as the profile {self._profile}."
 976                )
 977
 978            # Check if the replica profile exists in the configuration
 979            if replica_profile_name not in self._profiles:
 980                raise ValueError(f"Replica profile '{replica_profile_name}' not found in configuration")
 981
 982            # Get the replica profile configuration
 983            replica_profile_dict = self._profiles[replica_profile_name]
 984
 985            # Check if the replica profile has its own replicas configuration
 986            if replica_profile_dict.get("replicas"):
 987                raise ValueError(
 988                    f"Invalid replica configuration: profile '{replica_profile_name}' has its own replicas. "
 989                    f"This creates a circular reference which is not allowed."
 990                )
 991
 992    def build_config(self) -> "StorageClientConfig":
 993        bundle = self._provider_bundle
 994        backends = bundle.storage_backends
 995
 996        if len(backends) > 1:
 997            if not bundle.metadata_provider:
 998                raise ValueError(
 999                    f"Multi-backend configuration for profile '{self._profile}' requires metadata_provider "
1000                    "for routing between storage locations."
1001                )
1002
1003            child_profile_names = list(backends.keys())
1004            retry_config = self._build_retry_config()
1005            autocommit_config = self._build_autocommit_config()
1006
1007            # config for Composite StorageClient
1008            config = StorageClientConfig(
1009                profile=self._profile,
1010                storage_provider=None,
1011                credentials_provider=None,
1012                storage_provider_profiles=child_profile_names,
1013                child_configs=self._child_configs,
1014                metadata_provider=bundle.metadata_provider,
1015                cache_config=None,
1016                cache_manager=None,
1017                retry_config=retry_config,
1018                telemetry_provider=self._telemetry_provider,
1019                replicas=[],
1020                autocommit_config=autocommit_config,
1021            )
1022
1023            config._config_dict = self._resolved_config_dict
1024            return config
1025
1026        # Single-backend (len == 1)
1027        _, backend = next(iter(backends.items()))
1028        storage_provider = self._build_storage_provider(
1029            backend.storage_provider_config.type,
1030            backend.storage_provider_config.options,
1031            backend.credentials_provider,
1032        )
1033        credentials_provider = backend.credentials_provider
1034        metadata_provider = bundle.metadata_provider
1035        replicas = backend.replicas
1036
1037        # Validate replicas to prevent circular references
1038        if replicas:
1039            self._validate_replicas(replicas)
1040
1041        cache_config: CacheConfig | None = None
1042        cache_manager: CacheManager | None = None
1043
1044        # Check if caching is enabled for this profile
1045        caching_enabled = self._profile_dict.get("caching_enabled", False)
1046
1047        if self._cache_dict is not None and caching_enabled:
1048            tempdir = tempfile.gettempdir()
1049            default_location = os.path.join(tempdir, "msc_cache")
1050            location = self._cache_dict.get("location", default_location)
1051
1052            # Check if cache_backend.cache_path is defined
1053            cache_backend = self._cache_dict.get("cache_backend", {})
1054            cache_backend_path = cache_backend.get("cache_path") if cache_backend else None
1055
1056            # Warn if both location and cache_backend.cache_path are defined
1057            if cache_backend_path and self._cache_dict.get("location") is not None:
1058                logger.warning(
1059                    f"Both 'location' and 'cache_backend.cache_path' are defined in cache config. "
1060                    f"Using 'location' ({location}) and ignoring 'cache_backend.cache_path' ({cache_backend_path})."
1061                )
1062            elif cache_backend_path:
1063                # Use cache_backend.cache_path only if location is not explicitly defined
1064                location = cache_backend_path
1065
1066            check_source_version = self._cache_dict.get("check_source_version", True)
1067
1068            if not Path(location).is_absolute():
1069                raise ValueError(f"Cache location must be an absolute path: {location}")
1070
1071            # Initialize cache_dict with default values
1072            cache_dict = self._cache_dict
1073
1074            # Verify cache config
1075            self._verify_cache_config(cache_dict)
1076
1077            # Initialize eviction policy
1078            if "eviction_policy" in cache_dict:
1079                policy = cache_dict["eviction_policy"]["policy"].lower()
1080                purge_factor = cache_dict["eviction_policy"].get("purge_factor", 0)
1081                eviction_policy = EvictionPolicyConfig(
1082                    policy=policy,
1083                    refresh_interval=cache_dict["eviction_policy"].get(
1084                        "refresh_interval", DEFAULT_CACHE_REFRESH_INTERVAL
1085                    ),
1086                    purge_factor=purge_factor,
1087                )
1088            else:
1089                eviction_policy = EvictionPolicyConfig(policy="fifo", refresh_interval=DEFAULT_CACHE_REFRESH_INTERVAL)
1090
1091            # Create cache config from the standardized format
1092            cache_config = CacheConfig(
1093                size=cache_dict.get("size", DEFAULT_CACHE_SIZE),
1094                location=cache_dict.get("location", location),
1095                check_source_version=check_source_version,
1096                prefetch_file=cache_dict.get("prefetch_file", True),
1097                eviction_policy=eviction_policy,
1098                cache_line_size=cache_dict.get("cache_line_size", DEFAULT_CACHE_LINE_SIZE),
1099            )
1100
1101            cache_manager = CacheManager(profile=self._profile, cache_config=cache_config)
1102        elif self._cache_dict is not None and not caching_enabled:
1103            logger.debug(f"Caching is disabled for profile '{self._profile}'")
1104        elif self._cache_dict is None and caching_enabled:
1105            logger.warning(f"Caching is enabled for profile '{self._profile}' but no cache configuration is provided")
1106
1107        retry_config = self._build_retry_config()
1108        autocommit_config = self._build_autocommit_config()
1109
1110        config = StorageClientConfig(
1111            profile=self._profile,
1112            storage_provider=storage_provider,
1113            credentials_provider=credentials_provider,
1114            storage_provider_profiles=None,
1115            child_configs=None,
1116            metadata_provider=metadata_provider,
1117            cache_config=cache_config,
1118            cache_manager=cache_manager,
1119            retry_config=retry_config,
1120            telemetry_provider=self._telemetry_provider,
1121            replicas=replicas,
1122            autocommit_config=autocommit_config,
1123        )
1124
1125        config._config_dict = self._resolved_config_dict
1126        return config
1127
1128
1129class PathMapping:
1130    """
1131    Class to handle path mappings defined in the MSC configuration.
1132
1133    Path mappings create a nested structure of protocol -> bucket -> [(prefix, profile)]
1134    where entries are sorted by prefix length (longest first) for optimal matching.
1135    Longer paths take precedence when matching.
1136    """
1137
1138    def __init__(self):
1139        """Initialize an empty PathMapping."""
1140        self._mapping = defaultdict(lambda: defaultdict(list))
1141
1142    @classmethod
1143    def from_config(cls, config_dict: dict[str, Any] | None = None) -> "PathMapping":
1144        """
1145        Create a PathMapping instance from configuration dictionary.
1146
1147        :param config_dict: Configuration dictionary, if None the config will be loaded
1148        :return: A PathMapping instance with processed mappings
1149        """
1150        if config_dict is None:
1151            # Import locally to avoid circular imports
1152            from multistorageclient.config import StorageClientConfig
1153
1154            config_dict, _ = StorageClientConfig.read_msc_config()
1155
1156        if not config_dict:
1157            return cls()
1158
1159        instance = cls()
1160        instance._load_mapping(config_dict)
1161        return instance
1162
1163    def _load_mapping(self, config_dict: dict[str, Any]) -> None:
1164        """
1165        Load path mapping from a configuration dictionary.
1166
1167        :param config_dict: Configuration dictionary containing path mapping
1168        """
1169        # Get the path_mapping section
1170        path_mapping = config_dict.get("path_mapping", {})
1171        if path_mapping is None:
1172            return
1173
1174        # Process each mapping
1175        for source_path, dest_path in path_mapping.items():
1176            # Validate format
1177            if not source_path.endswith("/"):
1178                continue
1179            if not dest_path.startswith(MSC_PROTOCOL):
1180                continue
1181            if not dest_path.endswith("/"):
1182                continue
1183
1184            # Extract the destination profile
1185            pr_dest = urlparse(dest_path)
1186            dest_profile = pr_dest.netloc
1187
1188            # Parse the source path
1189            pr = urlparse(source_path)
1190            protocol = pr.scheme.lower() if pr.scheme else "file"
1191
1192            if protocol == "file" or source_path.startswith("/"):
1193                # For file or absolute paths, use the whole path as the prefix
1194                # and leave bucket empty
1195                bucket = ""
1196                prefix = source_path if source_path.startswith("/") else pr.path
1197            else:
1198                # For object storage, extract bucket and prefix
1199                bucket = pr.netloc
1200                prefix = pr.path
1201                prefix = prefix.removeprefix("/")
1202
1203            # Add the mapping to the nested dict
1204            self._mapping[protocol][bucket].append((prefix, dest_profile))
1205
1206        # Sort each bucket's prefixes by length (longest first) for optimal matching
1207        for protocol, buckets in self._mapping.items():
1208            for bucket, prefixes in buckets.items():
1209                self._mapping[protocol][bucket] = sorted(prefixes, key=lambda x: len(x[0]), reverse=True)
1210
1211    def find_mapping(self, url: str) -> tuple[str, str] | None:
1212        """
1213        Find the best matching mapping for the given URL.
1214
1215        :param url: URL to find matching mapping for
1216        :return: Tuple of (profile_name, translated_path) if a match is found, None otherwise
1217        """
1218        # Parse the URL
1219        pr = urlparse(url)
1220        protocol = pr.scheme.lower() if pr.scheme else "file"
1221
1222        # For file paths or absolute paths
1223        if protocol == "file" or url.startswith("/"):
1224            path = url if url.startswith("/") else pr.path
1225
1226            possible_mapping = self._mapping[protocol][""] if protocol in self._mapping else []
1227
1228            # Check each prefix (already sorted by length, longest first)
1229            for prefix, profile in possible_mapping:
1230                if path.startswith(prefix):
1231                    # Calculate the relative path
1232                    rel_path = path[len(prefix) :]
1233                    if not rel_path.startswith("/"):
1234                        rel_path = "/" + rel_path
1235                    return profile, rel_path
1236
1237            return None
1238
1239        # For object storage
1240        bucket = pr.netloc
1241        path = pr.path
1242        path = path.removeprefix("/")
1243
1244        # Check bucket-specific mapping
1245        possible_mapping = (
1246            self._mapping[protocol][bucket] if (protocol in self._mapping and bucket in self._mapping[protocol]) else []
1247        )
1248
1249        # Check each prefix (already sorted by length, longest first)
1250        for prefix, profile in possible_mapping:
1251            # matching prefix
1252            if path.startswith(prefix):
1253                rel_path = path[len(prefix) :]
1254                # Remove leading slash if present
1255                rel_path = rel_path.removeprefix("/")
1256
1257                return profile, rel_path
1258
1259        return None
1260
1261
[docs] 1262class StorageClientConfig: 1263 """ 1264 Configuration class for the :py:class:`multistorageclient.StorageClient`. 1265 """ 1266 1267 profile: str 1268 storage_provider: StorageProvider | None 1269 credentials_provider: CredentialsProvider | None 1270 storage_provider_profiles: list[str] | None 1271 child_configs: dict[str, "StorageClientConfig"] | None 1272 metadata_provider: MetadataProvider | None 1273 cache_config: CacheConfig | None 1274 cache_manager: CacheManager | None 1275 retry_config: RetryConfig | None 1276 telemetry_provider: Callable[[], Telemetry] | None 1277 replicas: list[Replica] 1278 autocommit_config: AutoCommitConfig | None 1279 1280 _config_dict: dict[str, Any] | None 1281 1282 def __init__( 1283 self, 1284 profile: str, 1285 storage_provider: StorageProvider | None = None, 1286 credentials_provider: CredentialsProvider | None = None, 1287 storage_provider_profiles: list[str] | None = None, 1288 child_configs: dict[str, "StorageClientConfig"] | None = None, 1289 metadata_provider: MetadataProvider | None = None, 1290 cache_config: CacheConfig | None = None, 1291 cache_manager: CacheManager | None = None, 1292 retry_config: RetryConfig | None = None, 1293 telemetry_provider: Callable[[], Telemetry] | None = None, 1294 replicas: list[Replica] | None = None, 1295 autocommit_config: AutoCommitConfig | None = None, 1296 ): 1297 # exactly one of storage_provider or storage_provider_profiles must be set 1298 if storage_provider and storage_provider_profiles: 1299 raise ValueError( 1300 "Cannot specify both storage_provider and storage_provider_profiles. " 1301 "Use storage_provider for SingleStorageClient or storage_provider_profiles for CompositeStorageClient." 1302 ) 1303 if not storage_provider and not storage_provider_profiles: 1304 raise ValueError("Must specify either storage_provider or storage_provider_profiles.") 1305 1306 if replicas is None: 1307 replicas = [] 1308 self.profile = profile 1309 self.storage_provider = storage_provider 1310 self.credentials_provider = credentials_provider 1311 self.storage_provider_profiles = storage_provider_profiles 1312 self.child_configs = child_configs 1313 self.metadata_provider = metadata_provider 1314 self.cache_config = cache_config 1315 self.cache_manager = cache_manager 1316 self.retry_config = retry_config 1317 self.telemetry_provider = telemetry_provider 1318 self.replicas = replicas 1319 self.autocommit_config = autocommit_config 1320
[docs] 1321 @staticmethod 1322 def from_json( 1323 config_json: str, 1324 profile: str = RESERVED_POSIX_PROFILE_NAME, 1325 telemetry_provider: Callable[[], Telemetry] | None = None, 1326 ) -> "StorageClientConfig": 1327 """ 1328 Load a storage client configuration from a JSON string. 1329 1330 :param config_json: Configuration JSON string. 1331 :param profile: Profile to use. 1332 :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. 1333 """ 1334 config_dict = json.loads(config_json) 1335 return StorageClientConfig.from_dict( 1336 config_dict=config_dict, profile=profile, telemetry_provider=telemetry_provider 1337 )
1338
[docs] 1339 @staticmethod 1340 def from_yaml( 1341 config_yaml: str, 1342 profile: str = RESERVED_POSIX_PROFILE_NAME, 1343 telemetry_provider: Callable[[], Telemetry] | None = None, 1344 ) -> "StorageClientConfig": 1345 """ 1346 Load a storage client configuration from a YAML string. 1347 1348 :param config_yaml: Configuration YAML string. 1349 :param profile: Profile to use. 1350 :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. 1351 """ 1352 config_dict = yaml.safe_load(config_yaml) or {} 1353 return StorageClientConfig.from_dict( 1354 config_dict=config_dict, profile=profile, telemetry_provider=telemetry_provider 1355 )
1356
[docs] 1357 @staticmethod 1358 def from_dict( 1359 config_dict: dict[str, Any], 1360 profile: str = RESERVED_POSIX_PROFILE_NAME, 1361 skip_validation: bool = False, 1362 telemetry_provider: Callable[[], Telemetry] | None = None, 1363 ) -> "StorageClientConfig": 1364 """ 1365 Load a storage client configuration from a Python dictionary. 1366 1367 :param config_dict: Configuration Python dictionary. 1368 :param profile: Profile to use. 1369 :param skip_validation: Skip configuration schema validation. 1370 :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. 1371 """ 1372 # Validate the config file with predefined JSON schema 1373 if not skip_validation: 1374 validate_config(config_dict) 1375 1376 # Load config 1377 loader = StorageClientConfigLoader( 1378 config_dict=config_dict, 1379 profile=profile, 1380 telemetry_provider=telemetry_provider, 1381 ) 1382 config = loader.build_config() 1383 1384 return config
1385
[docs] 1386 @staticmethod 1387 def from_file( 1388 config_file_paths: Iterable[str] | None = None, 1389 profile: str = RESERVED_POSIX_PROFILE_NAME, 1390 telemetry_provider: Callable[[], Telemetry] | None = None, 1391 ) -> "StorageClientConfig": 1392 """ 1393 Load a storage client configuration from the first file found. 1394 1395 :param config_file_paths: Configuration file search paths. If omitted, the default search paths are used (see :py:meth:`StorageClientConfig.read_msc_config`). 1396 :param profile: Profile to use. 1397 :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. 1398 """ 1399 msc_config_dict, msc_config_file = StorageClientConfig.read_msc_config(config_file_paths=config_file_paths) 1400 # Parse rclone config file. 1401 rclone_config_dict, rclone_config_file = read_rclone_config() 1402 1403 # Merge config files. 1404 merged_config, conflicted_keys = merge_dictionaries_no_overwrite(msc_config_dict, rclone_config_dict) 1405 if conflicted_keys: 1406 raise ValueError( 1407 f'Conflicting keys found in configuration files "{msc_config_file}" and "{rclone_config_file}: {conflicted_keys}' 1408 ) 1409 merged_profiles = merged_config.get("profiles", {}) 1410 1411 # Check if profile is in merged_profiles 1412 if profile in merged_profiles: 1413 return StorageClientConfig.from_dict( 1414 config_dict=merged_config, profile=profile, telemetry_provider=telemetry_provider 1415 ) 1416 else: 1417 # Check if profile is the default POSIX profile or an implicit profile 1418 if profile == RESERVED_POSIX_PROFILE_NAME: 1419 implicit_profile_config = DEFAULT_POSIX_PROFILE 1420 elif profile.startswith("_"): 1421 # Handle implicit profiles 1422 parts = profile[1:].split("-", 1) 1423 if len(parts) == 2: 1424 protocol, bucket = parts 1425 # Verify it's a supported protocol 1426 if protocol not in SUPPORTED_IMPLICIT_PROFILE_PROTOCOLS: 1427 raise ValueError(f'Unsupported protocol in implicit profile: "{protocol}"') 1428 implicit_profile_config = create_implicit_profile_config( 1429 profile_name=profile, protocol=protocol, base_path=bucket 1430 ) 1431 else: 1432 raise ValueError(f'Invalid implicit profile format: "{profile}"') 1433 else: 1434 raise ValueError( 1435 f'Profile "{profile}" not found in configuration files. Configuration was checked in ' 1436 f"{msc_config_file or 'MSC config (not found)'} and {rclone_config_file or 'Rclone config (not found)'}. " 1437 f"Please verify that the profile exists and that configuration files are correctly located." 1438 ) 1439 # merge the implicit profile config into the merged config so the cache & observability config can be inherited 1440 if "profiles" not in merged_config: 1441 merged_config["profiles"] = implicit_profile_config["profiles"] 1442 else: 1443 merged_config["profiles"][profile] = implicit_profile_config["profiles"][profile] 1444 # the config is already validated while reading, skip the validation for implicit profiles which start profile with "_" 1445 return StorageClientConfig.from_dict( 1446 config_dict=merged_config, profile=profile, skip_validation=True, telemetry_provider=telemetry_provider 1447 )
1448 1449 @staticmethod 1450 def from_provider_bundle( 1451 config_dict: dict[str, Any], 1452 provider_bundle: ProviderBundle | ProviderBundleV2, 1453 telemetry_provider: Callable[[], Telemetry] | None = None, 1454 ) -> "StorageClientConfig": 1455 loader = StorageClientConfigLoader( 1456 config_dict=config_dict, provider_bundle=provider_bundle, telemetry_provider=telemetry_provider 1457 ) 1458 config = loader.build_config() 1459 config._config_dict = None # Explicitly mark as None to avoid confusing pickling errors 1460 return config 1461
[docs] 1462 @staticmethod 1463 def read_msc_config( 1464 config_file_paths: Iterable[str] | None = None, 1465 ) -> tuple[dict[str, Any] | None, str | None]: 1466 """Get the MSC configuration dictionary and the path of the first file found. 1467 1468 If no config paths are specified, configs are searched in the following order: 1469 1470 1. ``MSC_CONFIG`` environment variable (highest precedence) 1471 2. Standard search paths (user-specified config and system-wide config) 1472 1473 :param config_file_paths: Configuration file search paths. If omitted, the default search paths are used. 1474 :return: Tuple of ``(config_dict, config_file_path)``. ``config_dict`` is the MSC configuration 1475 dictionary or empty dict if no config was found. ``config_file_path`` is the absolute 1476 path of the config file used, or ``None`` if no config file was found. 1477 """ 1478 config_dict: dict[str, Any] = {} 1479 config_file_path: str | None = None 1480 1481 config_file_paths = list(config_file_paths or []) 1482 1483 # Add default paths if none provided. 1484 if len(config_file_paths) == 0: 1485 # Environment variable. 1486 msc_config_env = os.getenv("MSC_CONFIG", None) 1487 if msc_config_env is not None: 1488 config_file_paths.append(msc_config_env) 1489 1490 # Standard search paths. 1491 config_file_paths.extend(_find_config_file_paths()) 1492 1493 # Normalize + absolutize paths. 1494 config_file_paths = [os.path.abspath(path) for path in config_file_paths] 1495 1496 # Log plan. 1497 logger.debug(f"Searching MSC config file paths: {config_file_paths}") 1498 1499 # Load config. 1500 for path in config_file_paths: 1501 if os.path.exists(path): 1502 try: 1503 with open(path) as f: 1504 if path.endswith(".json"): 1505 config_dict = json.load(f) 1506 else: 1507 config_dict = yaml.safe_load(f) 1508 config_file_path = path 1509 # Use the first config file. 1510 break 1511 except Exception as e: 1512 raise ValueError(f"malformed MSC config file: {path}, exception: {e}") 1513 1514 # Log result. 1515 if config_file_path is None: 1516 logger.debug("No MSC config files found in any of the search locations.") 1517 else: 1518 logger.debug(f"Using MSC config file: {config_file_path}") 1519 1520 if config_dict: 1521 validate_config(config_dict) 1522 1523 if "include" in config_dict and config_file_path: 1524 config_dict = _load_and_merge_includes(config_file_path, config_dict) 1525 1526 return config_dict, config_file_path
1527
[docs] 1528 @staticmethod 1529 def read_path_mapping() -> PathMapping: 1530 """ 1531 Get the path mapping defined in the MSC configuration. 1532 1533 Path mappings create a nested structure of protocol -> bucket -> [(prefix, profile)] 1534 where entries are sorted by prefix length (longest first) for optimal matching. 1535 Longer paths take precedence when matching. 1536 1537 :return: A PathMapping instance with translation mappings 1538 """ 1539 try: 1540 return PathMapping.from_config() 1541 except Exception: 1542 # Log the error but continue - this shouldn't stop the application from working 1543 logger.error("Failed to load path_mapping from MSC config") 1544 return PathMapping()
1545 1546 def __getstate__(self) -> dict[str, Any]: 1547 state = self.__dict__.copy() 1548 if not state.get("_config_dict"): 1549 raise ValueError("StorageClientConfig is not serializable") 1550 del state["credentials_provider"] 1551 del state["storage_provider"] 1552 del state["metadata_provider"] 1553 del state["cache_manager"] 1554 del state["replicas"] 1555 del state["child_configs"] 1556 return state 1557 1558 def __setstate__(self, state: dict[str, Any]) -> None: 1559 # Presence checked by __getstate__. 1560 config_dict = state["_config_dict"] 1561 loader = StorageClientConfigLoader( 1562 config_dict=config_dict, 1563 profile=state["profile"], 1564 telemetry_provider=state["telemetry_provider"], 1565 ) 1566 new_config = loader.build_config() 1567 self.profile = new_config.profile 1568 self.storage_provider = new_config.storage_provider 1569 self.credentials_provider = new_config.credentials_provider 1570 self.storage_provider_profiles = new_config.storage_provider_profiles 1571 self.child_configs = new_config.child_configs 1572 self.metadata_provider = new_config.metadata_provider 1573 self.cache_config = new_config.cache_config 1574 self.cache_manager = new_config.cache_manager 1575 self.retry_config = new_config.retry_config 1576 self.telemetry_provider = new_config.telemetry_provider 1577 self._config_dict = config_dict 1578 self.replicas = new_config.replicas 1579 self.autocommit_config = new_config.autocommit_config