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