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 )
794 )
795
796 # Sort replicas by read_priority
797 replicas.sort(key=lambda r: r.read_priority)
798
799 return SimpleProviderBundle(
800 storage_provider_config=StorageProviderConfig(storage_provider_name, storage_options),
801 credentials_provider=credentials_provider,
802 metadata_provider=metadata_provider,
803 replicas=replicas,
804 )
805
806 def _build_provider_bundle_from_extension(self, provider_bundle_dict: dict[str, Any]) -> ProviderBundle:
807 class_type = provider_bundle_dict["type"]
808 module_name, class_name = class_type.rsplit(".", 1)
809 cls = import_class(class_name, module_name)
810 options = provider_bundle_dict.get("options", {})
811 return cls(**options)
812
813 def _build_provider_bundle(self, provider_bundle: ProviderBundle | ProviderBundleV2 | None) -> ProviderBundleV2:
814 if provider_bundle:
815 bundle = provider_bundle
816 else:
817 provider_bundle_dict = self._profile_dict.get("provider_bundle", None)
818 if provider_bundle_dict:
819 bundle = self._build_provider_bundle_from_extension(provider_bundle_dict)
820 else:
821 bundle = self._build_provider_bundle_from_config(self._profile_dict)
822
823 if isinstance(bundle, ProviderBundle) and not isinstance(bundle, ProviderBundleV2):
824 bundle = SimpleProviderBundleV2.from_v1_bundle(self._profile, bundle)
825
826 return bundle
827
828 def _inject_profiles_from_bundle(self) -> None:
829 """
830 Inject child profiles and build child configs for multi-backend configurations.
831
832 For ProviderBundleV2 with multiple backends, this method:
833 1. Injects child profiles into config dict (needed for replica initialization)
834 2. Pre-builds child configs so CompositeStorageClient can use them directly
835
836 Profile injection is required because SingleStorageClient._initialize_replicas()
837 uses StorageClientConfig.from_dict() to create replica clients, which looks up
838 profiles in the config dict.
839 """
840 self._child_configs: dict[str, StorageClientConfig] | None = None
841
842 backends = self._provider_bundle.storage_backends
843 if len(backends) > 1:
844 # First, inject all child profiles into config dict (needed for replica lookup)
845 profiles = copy.deepcopy(self._profiles)
846 for child_name, backend in backends.items():
847 child_profile_dict = {
848 "storage_provider": {
849 "type": backend.storage_provider_config.type,
850 "options": backend.storage_provider_config.options,
851 }
852 }
853
854 if child_name in profiles:
855 # Profile already exists - check if it matches what we would inject
856 existing = profiles[child_name]
857 if existing.get("storage_provider") != child_profile_dict["storage_provider"]:
858 raise ValueError(
859 f"Profile '{child_name}' already exists in configuration with different settings."
860 )
861 else:
862 profiles[child_name] = child_profile_dict
863
864 # Update config dict BEFORE building child configs
865 resolved_config_dict = copy.deepcopy(self._resolved_config_dict)
866 resolved_config_dict["profiles"] = profiles
867 self._profiles = ImmutableDict(profiles)
868 self._resolved_config_dict = ImmutableDict(resolved_config_dict)
869
870 # Now build child configs (they will get the updated config dict)
871 retry_config = self._build_retry_config()
872 child_configs: dict[str, StorageClientConfig] = {}
873 for child_name, backend in backends.items():
874 storage_provider = self._build_storage_provider(
875 backend.storage_provider_config.type,
876 backend.storage_provider_config.options,
877 backend.credentials_provider,
878 )
879
880 child_config = StorageClientConfig(
881 profile=child_name,
882 storage_provider=storage_provider,
883 credentials_provider=backend.credentials_provider,
884 storage_provider_profiles=None,
885 child_configs=None,
886 metadata_provider=None,
887 cache_config=None,
888 cache_manager=None,
889 retry_config=retry_config,
890 telemetry_provider=self._telemetry_provider,
891 replicas=backend.replicas,
892 autocommit_config=None,
893 )
894 child_config._config_dict = self._resolved_config_dict
895 child_configs[child_name] = child_config
896
897 self._child_configs = child_configs
898
899 def _build_retry_config(self) -> RetryConfig:
900 """Build retry config from profile dict."""
901 retry_config_dict = self._profile_dict.get("retry", None)
902 if retry_config_dict:
903 attempts = retry_config_dict.get("attempts", DEFAULT_RETRY_ATTEMPTS)
904 delay = retry_config_dict.get("delay", DEFAULT_RETRY_DELAY)
905 backoff_multiplier = retry_config_dict.get("backoff_multiplier", DEFAULT_RETRY_BACKOFF_MULTIPLIER)
906 return RetryConfig(attempts=attempts, delay=delay, backoff_multiplier=backoff_multiplier)
907 else:
908 return RetryConfig(
909 attempts=DEFAULT_RETRY_ATTEMPTS,
910 delay=DEFAULT_RETRY_DELAY,
911 backoff_multiplier=DEFAULT_RETRY_BACKOFF_MULTIPLIER,
912 )
913
914 def _build_autocommit_config(self) -> AutoCommitConfig:
915 """Build autocommit config from profile dict."""
916 autocommit_dict = self._profile_dict.get("autocommit", None)
917 if autocommit_dict:
918 interval_minutes = autocommit_dict.get("interval_minutes", None)
919 at_exit = autocommit_dict.get("at_exit", False)
920 return AutoCommitConfig(interval_minutes=interval_minutes, at_exit=at_exit)
921 return AutoCommitConfig()
922
923 def _verify_cache_config(self, cache_dict: dict[str, Any]) -> None:
924 if "size_mb" in cache_dict:
925 raise ValueError(
926 "The 'size_mb' property is no longer supported. \n"
927 "Please use 'size' with a unit suffix (M, G, T) instead of size_mb.\n"
928 "Example configuration:\n"
929 "cache:\n"
930 " size: 500G # Optional: Maximum cache size (default: 10G)\n"
931 " cache_line_size: 64M # Optional: Chunk size for partial file caching (default: 64M)\n"
932 " check_source_version: true # Optional: Use ETag for cache validation (default: true)\n"
933 " location: /tmp/msc_cache # Optional: Cache directory path (default: system tempdir + '/msc_cache')\n"
934 " eviction_policy: # Optional: Cache eviction policy\n"
935 " policy: fifo # Optional: Policy type: lru, mru, fifo, random, no_eviction (default: fifo)\n"
936 " refresh_interval: 300 # Optional: Cache refresh interval in seconds (default: 300)\n"
937 )
938
939 # Validate that cache_line_size doesn't exceed cache size
940 cache_size_str = cache_dict.get("size", DEFAULT_CACHE_SIZE)
941 cache_line_size_str = cache_dict.get("cache_line_size", DEFAULT_CACHE_LINE_SIZE)
942
943 # Use CacheConfig to convert sizes to bytes for comparison
944 temp_cache_config = CacheConfig(
945 size=cache_size_str,
946 cache_line_size=cache_line_size_str,
947 )
948 cache_size_bytes = temp_cache_config.size_bytes()
949 cache_line_size_bytes = temp_cache_config.cache_line_size_bytes()
950
951 if cache_line_size_bytes > cache_size_bytes:
952 raise ValueError(
953 f"cache_line_size ({cache_line_size_str}) exceeds cache size ({cache_size_str}). "
954 f"cache_line_size must be less than or equal to cache size. "
955 f"Consider increasing cache size or decreasing cache_line_size."
956 )
957
958 def _validate_replicas(self, replicas: list[Replica]) -> None:
959 """
960 Validates that replica profiles do not have their own replicas configuration.
961
962 This prevents circular references where a replica profile could reference
963 another profile that also has replicas, creating an infinite loop.
964
965 :param replicas: The list of Replica objects to validate
966 :raises ValueError: If any replica profile has its own replicas configuration
967 """
968 for replica in replicas:
969 replica_profile_name = replica.replica_profile
970
971 # Check that replica profile is not the same as the current profile
972 if replica_profile_name == self._profile:
973 raise ValueError(
974 f"Replica profile {replica_profile_name} cannot be the same as the profile {self._profile}."
975 )
976
977 # Check if the replica profile exists in the configuration
978 if replica_profile_name not in self._profiles:
979 raise ValueError(f"Replica profile '{replica_profile_name}' not found in configuration")
980
981 # Get the replica profile configuration
982 replica_profile_dict = self._profiles[replica_profile_name]
983
984 # Check if the replica profile has its own replicas configuration
985 if replica_profile_dict.get("replicas"):
986 raise ValueError(
987 f"Invalid replica configuration: profile '{replica_profile_name}' has its own replicas. "
988 f"This creates a circular reference which is not allowed."
989 )
990
991 def build_config(self) -> "StorageClientConfig":
992 bundle = self._provider_bundle
993 backends = bundle.storage_backends
994
995 if len(backends) > 1:
996 if not bundle.metadata_provider:
997 raise ValueError(
998 f"Multi-backend configuration for profile '{self._profile}' requires metadata_provider "
999 "for routing between storage locations."
1000 )
1001
1002 child_profile_names = list(backends.keys())
1003 retry_config = self._build_retry_config()
1004 autocommit_config = self._build_autocommit_config()
1005
1006 # config for Composite StorageClient
1007 config = StorageClientConfig(
1008 profile=self._profile,
1009 storage_provider=None,
1010 credentials_provider=None,
1011 storage_provider_profiles=child_profile_names,
1012 child_configs=self._child_configs,
1013 metadata_provider=bundle.metadata_provider,
1014 cache_config=None,
1015 cache_manager=None,
1016 retry_config=retry_config,
1017 telemetry_provider=self._telemetry_provider,
1018 replicas=[],
1019 autocommit_config=autocommit_config,
1020 )
1021
1022 config._config_dict = self._resolved_config_dict
1023 return config
1024
1025 # Single-backend (len == 1)
1026 _, backend = next(iter(backends.items()))
1027 storage_provider = self._build_storage_provider(
1028 backend.storage_provider_config.type,
1029 backend.storage_provider_config.options,
1030 backend.credentials_provider,
1031 )
1032 credentials_provider = backend.credentials_provider
1033 metadata_provider = bundle.metadata_provider
1034 replicas = backend.replicas
1035
1036 # Validate replicas to prevent circular references
1037 if replicas:
1038 self._validate_replicas(replicas)
1039
1040 cache_config: CacheConfig | None = None
1041 cache_manager: CacheManager | None = None
1042
1043 # Check if caching is enabled for this profile
1044 caching_enabled = self._profile_dict.get("caching_enabled", False)
1045
1046 if self._cache_dict is not None and caching_enabled:
1047 tempdir = tempfile.gettempdir()
1048 default_location = os.path.join(tempdir, "msc_cache")
1049 location = self._cache_dict.get("location", default_location)
1050
1051 # Check if cache_backend.cache_path is defined
1052 cache_backend = self._cache_dict.get("cache_backend", {})
1053 cache_backend_path = cache_backend.get("cache_path") if cache_backend else None
1054
1055 # Warn if both location and cache_backend.cache_path are defined
1056 if cache_backend_path and self._cache_dict.get("location") is not None:
1057 logger.warning(
1058 f"Both 'location' and 'cache_backend.cache_path' are defined in cache config. "
1059 f"Using 'location' ({location}) and ignoring 'cache_backend.cache_path' ({cache_backend_path})."
1060 )
1061 elif cache_backend_path:
1062 # Use cache_backend.cache_path only if location is not explicitly defined
1063 location = cache_backend_path
1064
1065 check_source_version = self._cache_dict.get("check_source_version", True)
1066
1067 if not Path(location).is_absolute():
1068 raise ValueError(f"Cache location must be an absolute path: {location}")
1069
1070 # Initialize cache_dict with default values
1071 cache_dict = self._cache_dict
1072
1073 # Verify cache config
1074 self._verify_cache_config(cache_dict)
1075
1076 # Initialize eviction policy
1077 if "eviction_policy" in cache_dict:
1078 policy = cache_dict["eviction_policy"]["policy"].lower()
1079 purge_factor = cache_dict["eviction_policy"].get("purge_factor", 0)
1080 eviction_policy = EvictionPolicyConfig(
1081 policy=policy,
1082 refresh_interval=cache_dict["eviction_policy"].get(
1083 "refresh_interval", DEFAULT_CACHE_REFRESH_INTERVAL
1084 ),
1085 purge_factor=purge_factor,
1086 )
1087 else:
1088 eviction_policy = EvictionPolicyConfig(policy="fifo", refresh_interval=DEFAULT_CACHE_REFRESH_INTERVAL)
1089
1090 # Create cache config from the standardized format
1091 cache_config = CacheConfig(
1092 size=cache_dict.get("size", DEFAULT_CACHE_SIZE),
1093 location=cache_dict.get("location", location),
1094 check_source_version=check_source_version,
1095 prefetch_file=cache_dict.get("prefetch_file", True),
1096 eviction_policy=eviction_policy,
1097 cache_line_size=cache_dict.get("cache_line_size", DEFAULT_CACHE_LINE_SIZE),
1098 )
1099
1100 cache_manager = CacheManager(profile=self._profile, cache_config=cache_config)
1101 elif self._cache_dict is not None and not caching_enabled:
1102 logger.debug(f"Caching is disabled for profile '{self._profile}'")
1103 elif self._cache_dict is None and caching_enabled:
1104 logger.warning(f"Caching is enabled for profile '{self._profile}' but no cache configuration is provided")
1105
1106 retry_config = self._build_retry_config()
1107 autocommit_config = self._build_autocommit_config()
1108
1109 config = StorageClientConfig(
1110 profile=self._profile,
1111 storage_provider=storage_provider,
1112 credentials_provider=credentials_provider,
1113 storage_provider_profiles=None,
1114 child_configs=None,
1115 metadata_provider=metadata_provider,
1116 cache_config=cache_config,
1117 cache_manager=cache_manager,
1118 retry_config=retry_config,
1119 telemetry_provider=self._telemetry_provider,
1120 replicas=replicas,
1121 autocommit_config=autocommit_config,
1122 )
1123
1124 config._config_dict = self._resolved_config_dict
1125 return config
1126
1127
1128class PathMapping:
1129 """
1130 Class to handle path mappings defined in the MSC configuration.
1131
1132 Path mappings create a nested structure of protocol -> bucket -> [(prefix, profile)]
1133 where entries are sorted by prefix length (longest first) for optimal matching.
1134 Longer paths take precedence when matching.
1135 """
1136
1137 def __init__(self):
1138 """Initialize an empty PathMapping."""
1139 self._mapping = defaultdict(lambda: defaultdict(list))
1140
1141 @classmethod
1142 def from_config(cls, config_dict: dict[str, Any] | None = None) -> "PathMapping":
1143 """
1144 Create a PathMapping instance from configuration dictionary.
1145
1146 :param config_dict: Configuration dictionary, if None the config will be loaded
1147 :return: A PathMapping instance with processed mappings
1148 """
1149 if config_dict is None:
1150 # Import locally to avoid circular imports
1151 from multistorageclient.config import StorageClientConfig
1152
1153 config_dict, _ = StorageClientConfig.read_msc_config()
1154
1155 if not config_dict:
1156 return cls()
1157
1158 instance = cls()
1159 instance._load_mapping(config_dict)
1160 return instance
1161
1162 def _load_mapping(self, config_dict: dict[str, Any]) -> None:
1163 """
1164 Load path mapping from a configuration dictionary.
1165
1166 :param config_dict: Configuration dictionary containing path mapping
1167 """
1168 # Get the path_mapping section
1169 path_mapping = config_dict.get("path_mapping", {})
1170 if path_mapping is None:
1171 return
1172
1173 # Process each mapping
1174 for source_path, dest_path in path_mapping.items():
1175 # Validate format
1176 if not source_path.endswith("/"):
1177 continue
1178 if not dest_path.startswith(MSC_PROTOCOL):
1179 continue
1180 if not dest_path.endswith("/"):
1181 continue
1182
1183 # Extract the destination profile
1184 pr_dest = urlparse(dest_path)
1185 dest_profile = pr_dest.netloc
1186
1187 # Parse the source path
1188 pr = urlparse(source_path)
1189 protocol = pr.scheme.lower() if pr.scheme else "file"
1190
1191 if protocol == "file" or source_path.startswith("/"):
1192 # For file or absolute paths, use the whole path as the prefix
1193 # and leave bucket empty
1194 bucket = ""
1195 prefix = source_path if source_path.startswith("/") else pr.path
1196 else:
1197 # For object storage, extract bucket and prefix
1198 bucket = pr.netloc
1199 prefix = pr.path
1200 prefix = prefix.removeprefix("/")
1201
1202 # Add the mapping to the nested dict
1203 self._mapping[protocol][bucket].append((prefix, dest_profile))
1204
1205 # Sort each bucket's prefixes by length (longest first) for optimal matching
1206 for protocol, buckets in self._mapping.items():
1207 for bucket, prefixes in buckets.items():
1208 self._mapping[protocol][bucket] = sorted(prefixes, key=lambda x: len(x[0]), reverse=True)
1209
1210 def find_mapping(self, url: str) -> tuple[str, str] | None:
1211 """
1212 Find the best matching mapping for the given URL.
1213
1214 :param url: URL to find matching mapping for
1215 :return: Tuple of (profile_name, translated_path) if a match is found, None otherwise
1216 """
1217 # Parse the URL
1218 pr = urlparse(url)
1219 protocol = pr.scheme.lower() if pr.scheme else "file"
1220
1221 # For file paths or absolute paths
1222 if protocol == "file" or url.startswith("/"):
1223 path = url if url.startswith("/") else pr.path
1224
1225 possible_mapping = self._mapping[protocol][""] if protocol in self._mapping else []
1226
1227 # Check each prefix (already sorted by length, longest first)
1228 for prefix, profile in possible_mapping:
1229 if path.startswith(prefix):
1230 # Calculate the relative path
1231 rel_path = path[len(prefix) :]
1232 if not rel_path.startswith("/"):
1233 rel_path = "/" + rel_path
1234 return profile, rel_path
1235
1236 return None
1237
1238 # For object storage
1239 bucket = pr.netloc
1240 path = pr.path
1241 path = path.removeprefix("/")
1242
1243 # Check bucket-specific mapping
1244 possible_mapping = (
1245 self._mapping[protocol][bucket] if (protocol in self._mapping and bucket in self._mapping[protocol]) else []
1246 )
1247
1248 # Check each prefix (already sorted by length, longest first)
1249 for prefix, profile in possible_mapping:
1250 # matching prefix
1251 if path.startswith(prefix):
1252 rel_path = path[len(prefix) :]
1253 # Remove leading slash if present
1254 rel_path = rel_path.removeprefix("/")
1255
1256 return profile, rel_path
1257
1258 return None
1259
1260
[docs]
1261class StorageClientConfig:
1262 """
1263 Configuration class for the :py:class:`multistorageclient.StorageClient`.
1264 """
1265
1266 profile: str
1267 storage_provider: StorageProvider | None
1268 credentials_provider: CredentialsProvider | None
1269 storage_provider_profiles: list[str] | None
1270 child_configs: dict[str, "StorageClientConfig"] | None
1271 metadata_provider: MetadataProvider | None
1272 cache_config: CacheConfig | None
1273 cache_manager: CacheManager | None
1274 retry_config: RetryConfig | None
1275 telemetry_provider: Callable[[], Telemetry] | None
1276 replicas: list[Replica]
1277 autocommit_config: AutoCommitConfig | None
1278
1279 _config_dict: dict[str, Any] | None
1280
1281 def __init__(
1282 self,
1283 profile: str,
1284 storage_provider: StorageProvider | None = None,
1285 credentials_provider: CredentialsProvider | None = None,
1286 storage_provider_profiles: list[str] | None = None,
1287 child_configs: dict[str, "StorageClientConfig"] | None = None,
1288 metadata_provider: MetadataProvider | None = None,
1289 cache_config: CacheConfig | None = None,
1290 cache_manager: CacheManager | None = None,
1291 retry_config: RetryConfig | None = None,
1292 telemetry_provider: Callable[[], Telemetry] | None = None,
1293 replicas: list[Replica] | None = None,
1294 autocommit_config: AutoCommitConfig | None = None,
1295 ):
1296 # exactly one of storage_provider or storage_provider_profiles must be set
1297 if storage_provider and storage_provider_profiles:
1298 raise ValueError(
1299 "Cannot specify both storage_provider and storage_provider_profiles. "
1300 "Use storage_provider for SingleStorageClient or storage_provider_profiles for CompositeStorageClient."
1301 )
1302 if not storage_provider and not storage_provider_profiles:
1303 raise ValueError("Must specify either storage_provider or storage_provider_profiles.")
1304
1305 if replicas is None:
1306 replicas = []
1307 self.profile = profile
1308 self.storage_provider = storage_provider
1309 self.credentials_provider = credentials_provider
1310 self.storage_provider_profiles = storage_provider_profiles
1311 self.child_configs = child_configs
1312 self.metadata_provider = metadata_provider
1313 self.cache_config = cache_config
1314 self.cache_manager = cache_manager
1315 self.retry_config = retry_config
1316 self.telemetry_provider = telemetry_provider
1317 self.replicas = replicas
1318 self.autocommit_config = autocommit_config
1319
[docs]
1320 @staticmethod
1321 def from_json(
1322 config_json: str,
1323 profile: str = RESERVED_POSIX_PROFILE_NAME,
1324 telemetry_provider: Callable[[], Telemetry] | None = None,
1325 ) -> "StorageClientConfig":
1326 """
1327 Load a storage client configuration from a JSON string.
1328
1329 :param config_json: Configuration JSON string.
1330 :param profile: Profile to use.
1331 :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.
1332 """
1333 config_dict = json.loads(config_json)
1334 return StorageClientConfig.from_dict(
1335 config_dict=config_dict, profile=profile, telemetry_provider=telemetry_provider
1336 )
1337
[docs]
1338 @staticmethod
1339 def from_yaml(
1340 config_yaml: str,
1341 profile: str = RESERVED_POSIX_PROFILE_NAME,
1342 telemetry_provider: Callable[[], Telemetry] | None = None,
1343 ) -> "StorageClientConfig":
1344 """
1345 Load a storage client configuration from a YAML string.
1346
1347 :param config_yaml: Configuration YAML string.
1348 :param profile: Profile to use.
1349 :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.
1350 """
1351 config_dict = yaml.safe_load(config_yaml) or {}
1352 return StorageClientConfig.from_dict(
1353 config_dict=config_dict, profile=profile, telemetry_provider=telemetry_provider
1354 )
1355
[docs]
1356 @staticmethod
1357 def from_dict(
1358 config_dict: dict[str, Any],
1359 profile: str = RESERVED_POSIX_PROFILE_NAME,
1360 skip_validation: bool = False,
1361 telemetry_provider: Callable[[], Telemetry] | None = None,
1362 ) -> "StorageClientConfig":
1363 """
1364 Load a storage client configuration from a Python dictionary.
1365
1366 :param config_dict: Configuration Python dictionary.
1367 :param profile: Profile to use.
1368 :param skip_validation: Skip configuration schema validation.
1369 :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.
1370 """
1371 # Validate the config file with predefined JSON schema
1372 if not skip_validation:
1373 validate_config(config_dict)
1374
1375 # Load config
1376 loader = StorageClientConfigLoader(
1377 config_dict=config_dict,
1378 profile=profile,
1379 telemetry_provider=telemetry_provider,
1380 )
1381 config = loader.build_config()
1382
1383 return config
1384
[docs]
1385 @staticmethod
1386 def from_file(
1387 config_file_paths: Iterable[str] | None = None,
1388 profile: str = RESERVED_POSIX_PROFILE_NAME,
1389 telemetry_provider: Callable[[], Telemetry] | None = None,
1390 ) -> "StorageClientConfig":
1391 """
1392 Load a storage client configuration from the first file found.
1393
1394 :param config_file_paths: Configuration file search paths. If omitted, the default search paths are used (see :py:meth:`StorageClientConfig.read_msc_config`).
1395 :param profile: Profile to use.
1396 :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.
1397 """
1398 msc_config_dict, msc_config_file = StorageClientConfig.read_msc_config(config_file_paths=config_file_paths)
1399 # Parse rclone config file.
1400 rclone_config_dict, rclone_config_file = read_rclone_config()
1401
1402 # Merge config files.
1403 merged_config, conflicted_keys = merge_dictionaries_no_overwrite(msc_config_dict, rclone_config_dict)
1404 if conflicted_keys:
1405 raise ValueError(
1406 f'Conflicting keys found in configuration files "{msc_config_file}" and "{rclone_config_file}: {conflicted_keys}'
1407 )
1408 merged_profiles = merged_config.get("profiles", {})
1409
1410 # Check if profile is in merged_profiles
1411 if profile in merged_profiles:
1412 return StorageClientConfig.from_dict(
1413 config_dict=merged_config, profile=profile, telemetry_provider=telemetry_provider
1414 )
1415 else:
1416 # Check if profile is the default POSIX profile or an implicit profile
1417 if profile == RESERVED_POSIX_PROFILE_NAME:
1418 implicit_profile_config = DEFAULT_POSIX_PROFILE
1419 elif profile.startswith("_"):
1420 # Handle implicit profiles
1421 parts = profile[1:].split("-", 1)
1422 if len(parts) == 2:
1423 protocol, bucket = parts
1424 # Verify it's a supported protocol
1425 if protocol not in SUPPORTED_IMPLICIT_PROFILE_PROTOCOLS:
1426 raise ValueError(f'Unsupported protocol in implicit profile: "{protocol}"')
1427 implicit_profile_config = create_implicit_profile_config(
1428 profile_name=profile, protocol=protocol, base_path=bucket
1429 )
1430 else:
1431 raise ValueError(f'Invalid implicit profile format: "{profile}"')
1432 else:
1433 raise ValueError(
1434 f'Profile "{profile}" not found in configuration files. Configuration was checked in '
1435 f"{msc_config_file or 'MSC config (not found)'} and {rclone_config_file or 'Rclone config (not found)'}. "
1436 f"Please verify that the profile exists and that configuration files are correctly located."
1437 )
1438 # merge the implicit profile config into the merged config so the cache & observability config can be inherited
1439 if "profiles" not in merged_config:
1440 merged_config["profiles"] = implicit_profile_config["profiles"]
1441 else:
1442 merged_config["profiles"][profile] = implicit_profile_config["profiles"][profile]
1443 # the config is already validated while reading, skip the validation for implicit profiles which start profile with "_"
1444 return StorageClientConfig.from_dict(
1445 config_dict=merged_config, profile=profile, skip_validation=True, telemetry_provider=telemetry_provider
1446 )
1447
1448 @staticmethod
1449 def from_provider_bundle(
1450 config_dict: dict[str, Any],
1451 provider_bundle: ProviderBundle | ProviderBundleV2,
1452 telemetry_provider: Callable[[], Telemetry] | None = None,
1453 ) -> "StorageClientConfig":
1454 loader = StorageClientConfigLoader(
1455 config_dict=config_dict, provider_bundle=provider_bundle, telemetry_provider=telemetry_provider
1456 )
1457 config = loader.build_config()
1458 config._config_dict = None # Explicitly mark as None to avoid confusing pickling errors
1459 return config
1460
[docs]
1461 @staticmethod
1462 def read_msc_config(
1463 config_file_paths: Iterable[str] | None = None,
1464 ) -> tuple[dict[str, Any] | None, str | None]:
1465 """Get the MSC configuration dictionary and the path of the first file found.
1466
1467 If no config paths are specified, configs are searched in the following order:
1468
1469 1. ``MSC_CONFIG`` environment variable (highest precedence)
1470 2. Standard search paths (user-specified config and system-wide config)
1471
1472 :param config_file_paths: Configuration file search paths. If omitted, the default search paths are used.
1473 :return: Tuple of ``(config_dict, config_file_path)``. ``config_dict`` is the MSC configuration
1474 dictionary or empty dict if no config was found. ``config_file_path`` is the absolute
1475 path of the config file used, or ``None`` if no config file was found.
1476 """
1477 config_dict: dict[str, Any] = {}
1478 config_file_path: str | None = None
1479
1480 config_file_paths = list(config_file_paths or [])
1481
1482 # Add default paths if none provided.
1483 if len(config_file_paths) == 0:
1484 # Environment variable.
1485 msc_config_env = os.getenv("MSC_CONFIG", None)
1486 if msc_config_env is not None:
1487 config_file_paths.append(msc_config_env)
1488
1489 # Standard search paths.
1490 config_file_paths.extend(_find_config_file_paths())
1491
1492 # Normalize + absolutize paths.
1493 config_file_paths = [os.path.abspath(path) for path in config_file_paths]
1494
1495 # Log plan.
1496 logger.debug(f"Searching MSC config file paths: {config_file_paths}")
1497
1498 # Load config.
1499 for path in config_file_paths:
1500 if os.path.exists(path):
1501 try:
1502 with open(path) as f:
1503 if path.endswith(".json"):
1504 config_dict = json.load(f)
1505 else:
1506 config_dict = yaml.safe_load(f)
1507 config_file_path = path
1508 # Use the first config file.
1509 break
1510 except Exception as e:
1511 raise ValueError(f"malformed MSC config file: {path}, exception: {e}")
1512
1513 # Log result.
1514 if config_file_path is None:
1515 logger.debug("No MSC config files found in any of the search locations.")
1516 else:
1517 logger.debug(f"Using MSC config file: {config_file_path}")
1518
1519 if config_dict:
1520 validate_config(config_dict)
1521
1522 if "include" in config_dict and config_file_path:
1523 config_dict = _load_and_merge_includes(config_file_path, config_dict)
1524
1525 return config_dict, config_file_path
1526
[docs]
1527 @staticmethod
1528 def read_path_mapping() -> PathMapping:
1529 """
1530 Get the path mapping defined in the MSC configuration.
1531
1532 Path mappings create a nested structure of protocol -> bucket -> [(prefix, profile)]
1533 where entries are sorted by prefix length (longest first) for optimal matching.
1534 Longer paths take precedence when matching.
1535
1536 :return: A PathMapping instance with translation mappings
1537 """
1538 try:
1539 return PathMapping.from_config()
1540 except Exception:
1541 # Log the error but continue - this shouldn't stop the application from working
1542 logger.error("Failed to load path_mapping from MSC config")
1543 return PathMapping()
1544
1545 def __getstate__(self) -> dict[str, Any]:
1546 state = self.__dict__.copy()
1547 if not state.get("_config_dict"):
1548 raise ValueError("StorageClientConfig is not serializable")
1549 del state["credentials_provider"]
1550 del state["storage_provider"]
1551 del state["metadata_provider"]
1552 del state["cache_manager"]
1553 del state["replicas"]
1554 del state["child_configs"]
1555 return state
1556
1557 def __setstate__(self, state: dict[str, Any]) -> None:
1558 # Presence checked by __getstate__.
1559 config_dict = state["_config_dict"]
1560 loader = StorageClientConfigLoader(
1561 config_dict=config_dict,
1562 profile=state["profile"],
1563 telemetry_provider=state["telemetry_provider"],
1564 )
1565 new_config = loader.build_config()
1566 self.profile = new_config.profile
1567 self.storage_provider = new_config.storage_provider
1568 self.credentials_provider = new_config.credentials_provider
1569 self.storage_provider_profiles = new_config.storage_provider_profiles
1570 self.child_configs = new_config.child_configs
1571 self.metadata_provider = new_config.metadata_provider
1572 self.cache_config = new_config.cache_config
1573 self.cache_manager = new_config.cache_manager
1574 self.retry_config = new_config.retry_config
1575 self.telemetry_provider = new_config.telemetry_provider
1576 self._config_dict = config_dict
1577 self.replicas = new_config.replicas
1578 self.autocommit_config = new_config.autocommit_config