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