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 atexit
17import contextlib
18import logging
19import os
20import threading
21from collections.abc import Iterator, Sequence
22from concurrent.futures import ThreadPoolExecutor, as_completed
23from datetime import UTC, datetime
24from io import BytesIO
25from pathlib import PurePosixPath
26from typing import IO, Any, cast
27
28from ..config import StorageClientConfig
29from ..constants import DEFAULT_SYNC_BATCH_SIZE, MEMORY_LOAD_LIMIT
30from ..file import ObjectFile, PosixFile
31from ..providers.base import BaseStorageProvider
32from ..providers.posix_file import PosixFileStorageProvider
33from ..replica_manager import ReplicaManager
34from ..retry import batch_retry, retry
35from ..sync import SyncManager
36from ..types import (
37 AWARE_DATETIME_MIN,
38 MSC_PROTOCOL,
39 BatchTransferCallback,
40 ExecutionMode,
41 ObjectMetadata,
42 PatternList,
43 Range,
44 Replica,
45 ResolvedPathState,
46 SignerType,
47 SourceVersionCheckMode,
48 StorageProvider,
49 SuffixRange,
50 SymlinkHandling,
51 SyncResult,
52)
53from ..utils import (
54 NullStorageClient,
55 PatternMatcher,
56 _invoke_batch_transfer,
57 _transfer_notification,
58 create_attribute_filter_evaluator,
59 join_paths,
60 matches_attribute_filter_expression,
61)
62from .types import AbstractStorageClient
63
64logger = logging.getLogger(__name__)
65
66
[docs]
67class SingleStorageClient(AbstractStorageClient):
68 """
69 Storage client for single-backend configurations.
70
71 Supports full read and write operations against a single storage provider.
72 """
73
74 _config: StorageClientConfig
75 _storage_provider: StorageProvider
76 _metadata_provider_lock: "threading.Lock | None" = None
77 _stop_event: threading.Event | None = None
78 _replica_manager: ReplicaManager | None = None
79
80 def __init__(self, config: StorageClientConfig):
81 """
82 Initialize the :py:class:`SingleStorageClient` with the given configuration.
83
84 :param config: Storage client configuration with storage_provider set
85 :raises ValueError: If config has storage_provider_profiles (multi-backend)
86 """
87 self._initialize_providers(config)
88 self._initialize_replicas(config.replicas)
89
90 def _initialize_providers(self, config: StorageClientConfig) -> None:
91 if config.storage_provider_profiles:
92 raise ValueError(
93 "SingleStorageClient requires storage_provider, not storage_provider_profiles. "
94 "Use CompositeStorageClient for multi-backend configurations."
95 )
96
97 if config.storage_provider is None:
98 raise ValueError("SingleStorageClient requires storage_provider to be set.")
99
100 self._config = config
101 self._credentials_provider = self._config.credentials_provider
102 self._storage_provider = cast(StorageProvider, self._config.storage_provider)
103 self._metadata_provider = self._config.metadata_provider
104 self._cache_config = self._config.cache_config
105 self._retry_config = self._config.retry_config
106 self._cache_manager = self._config.cache_manager
107 self._autocommit_config = self._config.autocommit_config
108
109 if self._autocommit_config:
110 if self._metadata_provider:
111 logger.debug("Creating auto-commiter thread")
112
113 if self._autocommit_config.interval_minutes:
114 self._stop_event = threading.Event()
115 self._commit_thread = threading.Thread(
116 target=self._committer_thread,
117 daemon=True,
118 args=(self._autocommit_config.interval_minutes, self._stop_event),
119 )
120 self._commit_thread.start()
121
122 if self._autocommit_config.at_exit:
123 atexit.register(self._commit_on_exit)
124
125 self._metadata_provider_lock = threading.Lock()
126 else:
127 logger.debug("No metadata provider configured, auto-commit will not be enabled")
128
129 def _initialize_replicas(self, replicas: list[Replica]) -> None:
130 """Initialize replica StorageClient instances (facade)."""
131 # Import here to avoid circular dependency
132 from .client import StorageClient as StorageClientFacade
133
134 # Sort replicas by read_priority, the first one is the primary replica.
135 sorted_replicas = sorted(replicas, key=lambda r: r.read_priority)
136
137 replica_clients = []
138 writable_replica_clients = []
139 for replica in sorted_replicas:
140 if self._config._config_dict is None:
141 raise ValueError(f"Cannot initialize replica '{replica.replica_profile}' without a config")
142 replica_config = StorageClientConfig.from_dict(
143 config_dict=self._config._config_dict, profile=replica.replica_profile
144 )
145
146 storage_client = StorageClientFacade(config=replica_config)
147 replica_clients.append(storage_client)
148 if not replica.read_only:
149 writable_replica_clients.append(storage_client)
150
151 self._replicas = replica_clients
152 self._writable_replicas = writable_replica_clients
153 self._replica_manager = ReplicaManager(self) if len(self._replicas) > 0 else None
154
155 def _committer_thread(self, commit_interval_minutes: float, stop_event: threading.Event):
156 if not stop_event:
157 raise RuntimeError("Stop event not set")
158
159 while not stop_event.is_set():
160 # Wait with the ability to exit early
161 if stop_event.wait(timeout=commit_interval_minutes * 60):
162 break
163 logger.debug("Auto-committing to metadata provider")
164 self.commit_metadata()
165
166 def _commit_on_exit(self):
167 logger.debug("Shutting down, committing metadata one last time...")
168 self.commit_metadata()
169
170 def _get_source_version(self, path: str) -> str | None:
171 """
172 Get etag from metadata provider or storage provider.
173 """
174 if self._metadata_provider:
175 metadata = self._metadata_provider.get_object_metadata(path)
176 else:
177 metadata = self._storage_provider.get_object_metadata(path)
178 return metadata.etag
179
180 def _is_cache_enabled(self) -> bool:
181 enabled = self._cache_manager is not None and not self._is_posix_file_storage_provider()
182 return enabled
183
184 def _is_posix_file_storage_provider(self) -> bool:
185 """
186 :return: ``True`` if the storage client is using a POSIX file storage provider, ``False`` otherwise.
187 """
188 return isinstance(self._storage_provider, PosixFileStorageProvider)
189
190 def _is_rust_client_enabled(self) -> bool:
191 """
192 :return: ``True`` if the storage provider is using the Rust client, ``False`` otherwise.
193 """
194 return getattr(self._storage_provider, "_rust_client", None) is not None
195
196 def _get_suffix_object_size(self, physical_path: str) -> int:
197 """
198 :return: The size of the object at a physical path, following symlinks of the storage provider.
199 """
200 if isinstance(self._storage_provider, BaseStorageProvider):
201 provider = self._storage_provider
202 _, metadata = provider._resolve_symlink_path(provider._prepend_base_path(physical_path))
203 return metadata.content_length
204 return self._storage_provider.get_object_metadata(physical_path).content_length
205
206 def _can_read_suffix_range(self) -> bool:
207 """
208 :return: ``True`` if a suffix range can be passed to the storage provider without knowing the object size.
209 """
210 return (
211 isinstance(self._storage_provider, BaseStorageProvider)
212 and not (self._is_cache_enabled() and self._cache_manager)
213 and not self._replica_manager
214 )
215
216 def _read_from_replica_or_primary(self, path: str) -> bytes:
217 """
218 Read from replica or primary storage provider. Use BytesIO to avoid creating temporary files.
219 """
220 if self._replica_manager is None:
221 raise RuntimeError("Replica manager is not initialized")
222 file_obj = BytesIO()
223 self._replica_manager.download_from_replica_or_primary(path, file_obj, self._storage_provider)
224 return file_obj.getvalue()
225
226 def __del__(self):
227 if self._stop_event:
228 self._stop_event.set()
229 if self._commit_thread.is_alive():
230 self._commit_thread.join(timeout=5.0)
231
232 def __getstate__(self) -> dict[str, Any]:
233 state = self.__dict__.copy()
234 del state["_credentials_provider"]
235 del state["_storage_provider"]
236 del state["_metadata_provider"]
237 del state["_cache_manager"]
238
239 if "_metadata_provider_lock" in state:
240 del state["_metadata_provider_lock"]
241
242 if "_replicas" in state:
243 del state["_replicas"]
244 if "_writable_replicas" in state:
245 del state["_writable_replicas"]
246
247 # Replica manager could be disabled if it's set to None in the state.
248 if "_replica_manager" in state and state["_replica_manager"] is not None:
249 del state["_replica_manager"]
250
251 return state
252
253 def __setstate__(self, state: dict[str, Any]) -> None:
254 config = state["_config"]
255 self._initialize_providers(config)
256
257 # Replica manager could be disabled if it's set to None in the state.
258 if "_replica_manager" in state and state["_replica_manager"] is None:
259 self._replica_manager = None
260 else:
261 self._initialize_replicas(config.replicas)
262
263 if self._metadata_provider:
264 self._metadata_provider_lock = threading.Lock()
265
266 @property
267 def profile(self) -> str:
268 """
269 :return: The profile name of the storage client.
270 """
271 return self._config.profile
272
[docs]
273 def is_default_profile(self) -> bool:
274 """
275 :return: ``True`` if the storage client is using the reserved POSIX profile, ``False`` otherwise.
276 """
277 return self._config.profile == "__filesystem__"
278
279 @property
280 def replicas(self) -> list[AbstractStorageClient]:
281 """
282 :return: List of replica storage clients, sorted by read priority.
283 """
284 return self._replicas
285
286 @property
287 def writable_replicas(self) -> list[AbstractStorageClient]:
288 """:return: Writable replica storage clients, sorted by read priority."""
289 return self._writable_replicas
290
291 # -- Metadata resolution helpers --
292
293 def _resolve_read_path(self, logical_path: str) -> str:
294 """
295 Resolve a logical path to its physical storage path for read operations.
296
297 :param logical_path: The user-facing logical path.
298 :return: The physical storage path.
299 :raises FileNotFoundError: If the file does not exist in the metadata provider.
300 """
301 if self._metadata_provider is None:
302 raise RuntimeError("Metadata provider is not configured")
303 resolved = self._metadata_provider.realpath(logical_path)
304 if not resolved.exists:
305 raise FileNotFoundError(f"The file at path '{logical_path}' was not found by metadata provider.")
306 return resolved.physical_path
307
308 def _resolve_write_path(self, logical_path: str) -> str:
309 """
310 Resolve a logical path to its physical storage path for write operations.
311
312 Checks overwrite policy and generates the physical path via the metadata provider.
313
314 :param logical_path: The user-facing logical path.
315 :return: The physical storage path to write to.
316 :raises FileExistsError: If the file exists and overwrites are not allowed.
317 """
318 if self._metadata_provider is None:
319 raise RuntimeError("Metadata provider is not configured")
320 resolved = self._metadata_provider.realpath(logical_path)
321 if resolved.state in (ResolvedPathState.EXISTS, ResolvedPathState.DELETED):
322 if not self._metadata_provider.allow_overwrites():
323 raise FileExistsError(
324 f"The file at path '{logical_path}' already exists; "
325 f"overwriting is not allowed when using a metadata provider."
326 )
327 return self._metadata_provider.generate_physical_path(logical_path, for_overwrite=True).physical_path
328 return self._metadata_provider.generate_physical_path(logical_path, for_overwrite=False).physical_path
329
330 def _register_written_file(
331 self, virtual_path: str, physical_path: str, attributes: dict[str, Any] | None = None
332 ) -> None:
333 """
334 Register a written file with the metadata provider.
335
336 Fetches metadata from the storage provider, optionally merges custom attributes,
337 and registers the file with the metadata provider.
338
339 Protects metadata provider mutation with ``_metadata_provider_lock`` when configured.
340
341 .. note::
342 TODO(NGCDP-3016): Handle eventual consistency of Swiftstack, without wait.
343 """
344 if self._metadata_provider is None:
345 raise RuntimeError("Metadata provider is not configured")
346 obj_metadata = self._storage_provider.get_object_metadata(physical_path)
347 if attributes:
348 obj_metadata.metadata = (obj_metadata.metadata or {}) | attributes
349 with self._metadata_provider_lock or contextlib.nullcontext():
350 self._metadata_provider.add_file(virtual_path, obj_metadata)
351
352 def _register_written_files(
353 self,
354 virtual_paths: Sequence[str],
355 physical_paths: Sequence[str],
356 attributes: Sequence[dict[str, Any] | None] | None = None,
357 max_workers: int = 16,
358 ) -> None:
359 """Register multiple written files with the metadata provider concurrently."""
360 if not virtual_paths:
361 return
362
363 def _register_one(index: int, virtual_path: str, physical_path: str) -> None:
364 file_attrs = attributes[index] if attributes is not None else None
365 self._register_written_file(virtual_path, physical_path, file_attrs)
366
367 worker_count = max(1, min(len(virtual_paths), max_workers))
368 with ThreadPoolExecutor(max_workers=worker_count) as executor:
369 future_to_virtual_path = {
370 executor.submit(_register_one, i, virtual_path, physical_path): virtual_path
371 for i, (virtual_path, physical_path) in enumerate(zip(virtual_paths, physical_paths))
372 }
373 for future in as_completed(future_to_virtual_path):
374 future.result()
375
376 @batch_retry(operation_name="download_files")
377 def _download_files_batch(
378 self,
379 indices: list[int],
380 remote_paths: Sequence[str],
381 local_paths: Sequence[str],
382 metadata: Sequence[ObjectMetadata | None] | None,
383 max_workers: int,
384 transfer_callback: BatchTransferCallback | None = None,
385 ) -> None:
386 """
387 Execute one provider batch download attempt for the selected item indices.
388
389 The ``@batch_retry`` decorator retries this method with only failed
390 indices when the provider reports item-level retryable failures.
391
392 :param indices: Original batch indices to include in this attempt.
393 :param remote_paths: Remote paths for the full original batch.
394 :param local_paths: Local destination paths for the full original batch.
395 :param metadata: Optional per-file metadata for the full original batch.
396 :param max_workers: Maximum provider workers for this attempt.
397 :param transfer_callback: Optional observer called with source path, destination path, and "started",
398 "completed", or "failed" for each executing attempt. Remote paths may be resolved physical paths.
399 Calls may be concurrent; callback exceptions are logged.
400 """
401 provider_remote_paths = [remote_paths[index] for index in indices]
402 provider_local_paths = [local_paths[index] for index in indices]
403 provider_metadata = [metadata[index] for index in indices] if metadata is not None else None
404
405 _invoke_batch_transfer(
406 self._storage_provider.download_files,
407 provider_remote_paths,
408 provider_local_paths,
409 provider_metadata,
410 max_workers,
411 transfer_callback=transfer_callback,
412 )
413
414 @batch_retry(operation_name="upload_files")
415 def _upload_files_batch(
416 self,
417 indices: list[int],
418 local_paths: Sequence[str],
419 remote_paths: Sequence[str],
420 attributes: Sequence[dict[str, Any] | None] | None,
421 max_workers: int,
422 transfer_callback: BatchTransferCallback | None = None,
423 ) -> None:
424 """
425 Execute one provider batch upload attempt for the selected item indices.
426
427 The ``@batch_retry`` decorator retries this method with only failed
428 indices when the provider reports item-level retryable failures.
429
430 :param indices: Original batch indices to include in this attempt.
431 :param local_paths: Local source paths for the full original batch.
432 :param remote_paths: Provider destination paths for the full original batch.
433 :param attributes: Optional per-file attributes for the full original batch.
434 :param max_workers: Maximum provider workers for this attempt.
435 :param transfer_callback: Optional observer called with source path, destination path, and "started",
436 "completed", or "failed" for each executing attempt. Remote paths may be resolved physical paths.
437 Calls may be concurrent; callback exceptions are logged.
438 """
439 provider_local_paths = [local_paths[index] for index in indices]
440 provider_remote_paths = [remote_paths[index] for index in indices]
441 provider_attributes = [attributes[index] for index in indices] if attributes is not None else None
442
443 _invoke_batch_transfer(
444 self._storage_provider.upload_files,
445 provider_local_paths,
446 provider_remote_paths,
447 provider_attributes,
448 max_workers,
449 transfer_callback=transfer_callback,
450 )
451
[docs]
452 @retry
453 def read(
454 self,
455 path: str,
456 byte_range: Range | SuffixRange | None = None,
457 check_source_version: SourceVersionCheckMode = SourceVersionCheckMode.INHERIT,
458 ) -> bytes:
459 """
460 Read bytes from a file at the specified logical path.
461
462 :param path: The logical path of the object to read.
463 :param byte_range: Optional byte range to read (offset and length), or a suffix range to read the last bytes.
464 :param check_source_version: Whether to check the source version of cached objects.
465 :return: The content of the object as bytes.
466 :raises FileNotFoundError: If the file at the specified path does not exist.
467 """
468 if self._metadata_provider:
469 path = self._resolve_read_path(path)
470
471 if isinstance(byte_range, SuffixRange):
472 if self._can_read_suffix_range():
473 return cast(BaseStorageProvider, self._storage_provider).get_object(path, byte_range=byte_range)
474
475 # (the size of the physical object: the path is resolved, symlinks of the storage provider are followed)
476 content_length = self._get_suffix_object_size(path)
477 offset = max(content_length - byte_range.size, 0)
478 if offset >= content_length:
479 return b""
480 byte_range = Range(offset=offset, size=content_length - offset)
481
482 # Handle caching logic
483 if self._is_cache_enabled() and self._cache_manager:
484 if byte_range:
485 # Range request with cache
486 try:
487 # Fetch metadata for source version checking (if needed)
488 metadata = None
489 source_version = None
490 if check_source_version == SourceVersionCheckMode.ENABLE:
491 metadata = self._storage_provider.get_object_metadata(path)
492 source_version = metadata.etag
493 elif check_source_version == SourceVersionCheckMode.INHERIT:
494 if self._cache_manager.check_source_version():
495 metadata = self._storage_provider.get_object_metadata(path)
496 source_version = metadata.etag
497
498 # Optimization: For full-file reads (offset=0, size >= file_size), cache whole file instead of chunking
499 # This avoids creating many small chunks when the user requests the entire file.
500 # Only apply this optimization when metadata is already available (i.e., when version checking is enabled),
501 # to respect the user's choice to disable version checking and avoid extra HEAD requests.
502 if byte_range.offset == 0 and metadata and byte_range.size >= metadata.content_length:
503 full_file_data = self._cache_manager.read(path, source_version)
504 if full_file_data is None:
505 if self._replica_manager:
506 full_file_data = self._read_from_replica_or_primary(path)
507 else:
508 full_file_data = self._storage_provider.get_object(path)
509 self._cache_manager.set(path, full_file_data, source_version)
510 return full_file_data[: metadata.content_length]
511
512 # Use chunk-based caching for partial reads or when optimization doesn't apply
513 data = self._cache_manager.read(
514 key=path,
515 source_version=source_version,
516 byte_range=byte_range,
517 storage_provider=self._storage_provider,
518 source_size=metadata.content_length if metadata else None,
519 check_source_version=check_source_version,
520 )
521 if data is not None:
522 return data
523 # Fallback (should not normally happen)
524 return self._storage_provider.get_object(path, byte_range=byte_range)
525 except (FileNotFoundError, Exception):
526 # Fall back to direct read if metadata fetching fails
527 return self._storage_provider.get_object(path, byte_range=byte_range)
528 else:
529 # Full file read with cache
530 # Only fetch source version if check_source_version is enabled
531 source_version = None
532 if check_source_version == SourceVersionCheckMode.ENABLE or (
533 check_source_version == SourceVersionCheckMode.INHERIT
534 and self._cache_manager.check_source_version()
535 ):
536 source_version = self._get_source_version(path)
537
538 data = self._cache_manager.read(path, source_version)
539 if data is None:
540 if self._replica_manager:
541 data = self._read_from_replica_or_primary(path)
542 else:
543 data = self._storage_provider.get_object(path)
544 self._cache_manager.set(path, data, source_version)
545 return data
546 elif self._replica_manager:
547 # No cache, but replicas available
548 data = self._read_from_replica_or_primary(path)
549 if byte_range is not None:
550 return data[byte_range.offset : byte_range.offset + byte_range.size]
551 return data
552 else:
553 # No cache, no replicas - direct storage provider read
554 return self._storage_provider.get_object(path, byte_range=byte_range)
555
[docs]
556 def info(self, path: str, strict: bool = True) -> ObjectMetadata:
557 """
558 Get metadata for a file at the specified path.
559
560 :param path: The logical path of the object.
561 :param strict: When ``True``, only return committed metadata. When ``False``, include pending changes.
562 :return: ObjectMetadata containing file information (size, last modified, etc.).
563 :raises FileNotFoundError: If the file at the specified path does not exist.
564 """
565 if not path or path == ".": # empty path or '.' provided by the user
566 if self._is_posix_file_storage_provider():
567 root = cast(PosixFileStorageProvider, self._storage_provider)._prepend_base_path("")
568 last_modified = datetime.fromtimestamp(os.path.getmtime(root), tz=UTC)
569 else:
570 last_modified = AWARE_DATETIME_MIN
571 return ObjectMetadata(key="", type="directory", content_length=0, last_modified=last_modified)
572
573 if not self._metadata_provider:
574 return self._storage_provider.get_object_metadata(path, strict=strict)
575
576 return self._metadata_provider.get_object_metadata(path, include_pending=not strict)
577
[docs]
578 @retry
579 def download_file(self, remote_path: str, local_path: str | IO, metadata: ObjectMetadata | None = None) -> None:
580 """
581 Download a remote file to a local path or file-like object.
582
583 :param remote_path: The logical path of the remote file to download.
584 :param local_path: The local file path or file-like object to write to.
585 :param metadata: Optional metadata for this object, as returned by :py:meth:`info`, to avoid a provider lookup.
586 :raises FileNotFoundError: If the remote file does not exist.
587 """
588 if self._metadata_provider:
589 physical_path = self._resolve_read_path(remote_path)
590 metadata = self._metadata_provider.get_object_metadata(remote_path)
591 self._storage_provider.download_file(physical_path, local_path, metadata)
592 elif self._replica_manager:
593 self._replica_manager.download_from_replica_or_primary(remote_path, local_path, self._storage_provider)
594 else:
595 self._storage_provider.download_file(remote_path, local_path, metadata)
596
[docs]
597 def download_files(
598 self,
599 remote_paths: Sequence[str],
600 local_paths: Sequence[str],
601 metadata: Sequence[ObjectMetadata | None] | None = None,
602 max_workers: int = 16,
603 transfer_callback: BatchTransferCallback | None = None,
604 ) -> None:
605 """
606 Download multiple remote files to local paths.
607
608 :param remote_paths: List of logical paths of remote files to download.
609 :param local_paths: List of local file paths to save the downloaded files to.
610 :param metadata: Optional per-file metadata used to decide between regular and multipart download.
611 :param max_workers: Maximum number of concurrent download workers (default: 16).
612 :param transfer_callback: Optional observer called with source path, destination path, and "started",
613 "completed", or "failed" for each executing attempt. Remote paths may be resolved physical paths.
614 Calls may be concurrent; callback exceptions are logged.
615 :raises ValueError: If remote_paths and local_paths have different lengths.
616 :raises FileNotFoundError: If any remote file does not exist.
617 """
618 if len(remote_paths) != len(local_paths):
619 raise ValueError("remote_paths and local_paths must have the same length")
620
621 if metadata is not None and len(metadata) != len(remote_paths):
622 raise ValueError("metadata must have the same length as remote_paths and local_paths")
623
624 if self._metadata_provider:
625 physical_paths = [self._resolve_read_path(rp) for rp in remote_paths]
626 self._download_files_batch(
627 list(range(len(remote_paths))), physical_paths, local_paths, metadata, max_workers, transfer_callback
628 )
629 elif self._replica_manager:
630 for remote_path, local_path in zip(remote_paths, local_paths):
631 with _transfer_notification(transfer_callback, remote_path, local_path):
632 self.download_file(remote_path, local_path)
633 else:
634 self._download_files_batch(
635 list(range(len(remote_paths))), remote_paths, local_paths, metadata, max_workers, transfer_callback
636 )
637
[docs]
638 @retry
639 def upload_file(self, remote_path: str, local_path: str | IO, attributes: dict[str, Any] | None = None) -> None:
640 """
641 Uploads a file from the local file system to the storage provider.
642
643 :param remote_path: The path where the object will be stored.
644 :param local_path: The source file to upload. This can either be a string representing the local
645 file path, or a file-like object (e.g., an open file handle).
646 :param attributes: The attributes to add to the file if a new file is created.
647 """
648 virtual_path = remote_path
649 if self._metadata_provider:
650 physical_path = self._resolve_write_path(remote_path)
651 # Attributes belong to logical metadata, so physical writes get none and registration stores them.
652 self._storage_provider.upload_file(physical_path, local_path, attributes=None)
653 self._register_written_file(virtual_path, physical_path, attributes)
654 else:
655 self._storage_provider.upload_file(remote_path, local_path, attributes)
656
[docs]
657 def upload_files(
658 self,
659 remote_paths: Sequence[str],
660 local_paths: Sequence[str],
661 attributes: Sequence[dict[str, Any] | None] | None = None,
662 max_workers: int = 16,
663 transfer_callback: BatchTransferCallback | None = None,
664 ) -> None:
665 """
666 Upload multiple local files to remote storage.
667
668 :param remote_paths: List of logical paths where the files will be uploaded.
669 :param local_paths: List of local file paths to upload.
670 :param attributes: Optional list of per-file attributes to add. When provided, must have the same length
671 as remote_paths/local_paths. Each element may be ``None`` for files that need no attributes.
672 :param max_workers: Maximum number of concurrent upload workers (default: 16).
673 :param transfer_callback: Optional observer called with source path, destination path, and "started",
674 "completed", or "failed" for each executing attempt. Remote paths may be resolved physical paths.
675 Calls may be concurrent; callback exceptions are logged.
676 :raises ValueError: If remote_paths and local_paths have different lengths.
677 :raises ValueError: If attributes is provided and has a different length than remote_paths.
678 """
679 if len(remote_paths) != len(local_paths):
680 raise ValueError("remote_paths and local_paths must have the same length")
681
682 if attributes is not None and len(attributes) != len(remote_paths):
683 raise ValueError("attributes must have the same length as remote_paths and local_paths")
684
685 if self._metadata_provider:
686 physical_paths = [self._resolve_write_path(rp) for rp in remote_paths]
687
688 # Attributes belong to logical metadata, so physical writes get none and registration stores them.
689 self._upload_files_batch(
690 list(range(len(remote_paths))),
691 local_paths,
692 physical_paths,
693 None,
694 max_workers,
695 transfer_callback,
696 )
697
698 self._register_written_files(remote_paths, physical_paths, attributes, max_workers)
699 else:
700 self._upload_files_batch(
701 list(range(len(remote_paths))), local_paths, remote_paths, attributes, max_workers, transfer_callback
702 )
703
[docs]
704 @retry
705 def write(self, path: str, body: bytes, attributes: dict[str, Any] | None = None) -> None:
706 """
707 Write bytes to a file at the specified path.
708
709 :param path: The logical path where the object will be written.
710 :param body: The content to write as bytes.
711 :param attributes: Optional attributes to add to the file.
712 """
713 virtual_path = path
714 if self._metadata_provider:
715 physical_path = self._resolve_write_path(path)
716 # Attributes belong to logical metadata, so physical writes get none and registration stores them.
717 self._storage_provider.put_object(physical_path, body, attributes=None)
718 self._register_written_file(virtual_path, physical_path, attributes)
719 else:
720 self._storage_provider.put_object(path, body, attributes=attributes)
721
[docs]
722 def copy(self, src_path: str, dest_path: str) -> None:
723 """
724 Copy a file from source path to destination path.
725
726 :param src_path: The logical path of the source object.
727 :param dest_path: The logical path where the object will be copied to.
728 :raises FileNotFoundError: If the source file does not exist.
729 """
730 virtual_dest_path = dest_path
731 if self._metadata_provider:
732 src_path = self._resolve_read_path(src_path)
733 dest_path = self._resolve_write_path(dest_path)
734 self._storage_provider.copy_object(src_path, dest_path)
735 self._register_written_file(virtual_dest_path, dest_path)
736 else:
737 self._storage_provider.copy_object(src_path, dest_path)
738
[docs]
739 def make_symlink(self, path: str, target: str) -> None:
740 """
741 Creates a symbolic link at ``path`` pointing to ``target``.
742
743 :param path: The logical path where the symlink will be created.
744 :param target: The logical key that the symlink points to.
745 """
746 if self._metadata_provider:
747 try:
748 self._metadata_provider.get_object_metadata(path)
749 if not self._metadata_provider.allow_overwrites():
750 raise FileExistsError(
751 f"The file at path '{path}' already exists; "
752 f"overwriting is not allowed when using a metadata provider."
753 )
754 except FileNotFoundError:
755 pass
756 obj_metadata = ObjectMetadata(
757 key=path,
758 content_length=0,
759 last_modified=datetime.now(tz=UTC),
760 symlink_target=ObjectMetadata.encode_symlink_target(path, target),
761 )
762 with self._metadata_provider_lock or contextlib.nullcontext():
763 self._metadata_provider.add_file(path, obj_metadata)
764 else:
765 self._storage_provider.make_symlink(path, target)
766
[docs]
767 def delete(self, path: str, recursive: bool = False) -> None:
768 """
769 Deletes an object at the specified path.
770
771 :param path: The logical path of the object or directory to delete.
772 :param recursive: Whether to delete objects in the path recursively.
773 """
774 obj_metadata = self.info(path)
775 is_dir = obj_metadata and obj_metadata.type == "directory"
776 is_file = obj_metadata and obj_metadata.type == "file"
777 if recursive and is_dir:
778 self.sync_from(
779 cast(AbstractStorageClient, NullStorageClient()),
780 path,
781 path,
782 delete_unmatched_files=True,
783 num_worker_processes=1,
784 description="Deleting",
785 )
786 # If this is a posix storage provider, we need to also delete remaining directory stubs.
787 # TODO: Notify metadata provider of the changes.
788 if self._is_posix_file_storage_provider():
789 posix_storage_provider = cast(PosixFileStorageProvider, self._storage_provider)
790 posix_storage_provider.rmtree(path)
791 return
792 else:
793 # 1) If path is a file: delete the file
794 # 2) If path is a directory: raise an error to prompt the user to use the recursive flag
795 if is_file:
796 virtual_path = path
797 if self._metadata_provider:
798 resolved = self._metadata_provider.realpath(path)
799 if not resolved.exists:
800 raise FileNotFoundError(f"The file at path '{virtual_path}' was not found.")
801
802 # Check if soft-delete is enabled
803 if not self._metadata_provider.should_use_soft_delete():
804 # Hard delete: remove both physical file and metadata
805 self._storage_provider.delete_object(resolved.physical_path)
806
807 with self._metadata_provider_lock or contextlib.nullcontext():
808 self._metadata_provider.remove_file(virtual_path)
809 else:
810 self._storage_provider.delete_object(path)
811
812 # Delete the cached file if it exists
813 if self._is_cache_enabled():
814 if self._cache_manager is None:
815 raise RuntimeError("Cache manager is not initialized")
816 self._cache_manager.delete(virtual_path)
817
818 # Delete from replicas if replica manager exists
819 if self._replica_manager:
820 self._replica_manager.delete_from_replicas(virtual_path)
821 elif is_dir:
822 raise ValueError(f"'{path}' is a directory. Set recursive=True to delete entire directory.")
823 else:
824 raise FileNotFoundError(f"The file at '{path}' was not found.")
825
[docs]
826 @retry
827 def delete_many(self, paths: Sequence[str]) -> None:
828 """
829 Delete multiple files at the specified paths. Only files are supported; directories are not deleted.
830
831 :param paths: List of logical paths of the files to delete.
832 """
833 physical_paths_to_delete: list[str] = []
834 metadata_paths_to_remove: set[str] = set()
835 for path in paths:
836 if self._metadata_provider:
837 resolved = self._metadata_provider.realpath(path)
838 if not resolved.exists:
839 continue
840 metadata_paths_to_remove.add(path)
841 if not self._metadata_provider.should_use_soft_delete():
842 physical_paths_to_delete.append(resolved.physical_path)
843 else:
844 physical_paths_to_delete.append(path)
845
846 if physical_paths_to_delete:
847 self._storage_provider.delete_objects(physical_paths_to_delete)
848
849 for path in paths:
850 virtual_path = path
851 if self._metadata_provider and virtual_path in metadata_paths_to_remove:
852 with self._metadata_provider_lock or contextlib.nullcontext():
853 self._metadata_provider.remove_file(virtual_path)
854 if self._is_cache_enabled():
855 if self._cache_manager is None:
856 raise RuntimeError("Cache manager is not initialized")
857 self._cache_manager.delete(virtual_path)
858 if self._replica_manager:
859 self._replica_manager.delete_from_replicas(virtual_path)
860
[docs]
861 def glob(
862 self,
863 pattern: str,
864 include_url_prefix: bool = False,
865 attribute_filter_expression: str | None = None,
866 ) -> list[str]:
867 """
868 Matches and retrieves a list of object keys in the storage provider that match the specified pattern.
869
870 :param pattern: The pattern to match object keys against, supporting wildcards (e.g., ``*.txt``).
871 :param include_url_prefix: Whether to include the URL prefix ``msc://profile`` in the result.
872 :param attribute_filter_expression: The attribute filter expression to apply to the result.
873 :return: A list of object paths that match the specified pattern.
874 """
875 if self._metadata_provider:
876 results = self._metadata_provider.glob(pattern, attribute_filter_expression)
877 else:
878 results = self._storage_provider.glob(pattern, attribute_filter_expression)
879
880 if include_url_prefix:
881 results = [join_paths(f"{MSC_PROTOCOL}{self._config.profile}", path) for path in results]
882
883 return results
884
885 def _resolve_single_file(
886 self,
887 path: str,
888 start_after: str | None,
889 end_at: str | None,
890 include_url_prefix: bool,
891 pattern_matcher: PatternMatcher | None,
892 ) -> tuple[ObjectMetadata | None, str | None]:
893 """
894 Resolve whether ``path`` should be handled as a single-file listing result.
895
896 :param path: Candidate file path or directory prefix to resolve.
897 :param start_after: Exclusive lower bound for file key filtering.
898 :param end_at: Inclusive upper bound for file key filtering.
899 :param include_url_prefix: Whether to prefix returned keys with ``msc://profile``.
900 :param pattern_matcher: Optional include/exclude matcher for file keys.
901 :return: A tuple of ``(single_file, normalized_path)``. Returns file metadata and
902 the original path when ``path`` resolves to a file that passes filters;
903 returns ``(None, normalized_directory_path)`` when the caller should
904 continue with directory listing; returns ``(None, None)`` when filtering
905 excludes the single-file candidate and listing should stop.
906 """
907 if not path:
908 return None, path
909
910 if self._is_posix_file_storage_provider() and not self._metadata_provider:
911 provider = cast(PosixFileStorageProvider, self._storage_provider)
912 normalized_path = path.rstrip("/") or path
913 physical_path = provider._prepend_base_path(normalized_path)
914 if os.path.islink(physical_path):
915 return None, normalized_path
916
917 if self.is_file(path):
918 if pattern_matcher and not pattern_matcher.should_include_file(path):
919 return None, None
920
921 try:
922 object_metadata = self.info(path)
923 if start_after and object_metadata.key <= start_after:
924 return None, None
925 if end_at and object_metadata.key > end_at:
926 return None, None
927 if include_url_prefix:
928 object_metadata = self._with_url_prefix(object_metadata)
929 return object_metadata, path
930 except FileNotFoundError:
931 return None, path.rstrip("/") + "/"
932 else:
933 return None, path.rstrip("/") + "/"
934
935 def _with_url_prefix(self, obj: ObjectMetadata) -> ObjectMetadata:
936 # Return a copy: listed objects may be owned by the metadata provider and must not be mutated.
937 if self.is_default_profile():
938 return obj.replace(key=str(PurePosixPath("/") / obj.key))
939 return obj.replace(key=join_paths(f"{MSC_PROTOCOL}{self._config.profile}", obj.key))
940
941 def _filter_and_decorate(
942 self,
943 objects: Iterator[ObjectMetadata],
944 include_url_prefix: bool,
945 pattern_matcher: PatternMatcher | None,
946 ) -> Iterator[ObjectMetadata]:
947 for obj in objects:
948 if pattern_matcher and not pattern_matcher.should_include_file(obj.key):
949 continue
950 if include_url_prefix:
951 obj = self._with_url_prefix(obj)
952 yield obj
953
[docs]
954 def list_recursive(
955 self,
956 path: str = "",
957 start_after: str | None = None,
958 end_at: str | None = None,
959 max_workers: int = 32,
960 look_ahead: int = 2,
961 include_url_prefix: bool = False,
962 patterns: PatternList | None = None,
963 symlink_handling: SymlinkHandling = SymlinkHandling.FOLLOW,
964 ) -> Iterator[ObjectMetadata]:
965 """
966 List files recursively in the storage provider under the specified path.
967
968 :param path: The directory or file path to list objects under. This should be a
969 complete filesystem path (e.g., "my-bucket/documents/" or "data/2024/").
970 :param start_after: The key to start after (i.e. exclusive). An object with this key doesn't have to exist.
971 :param end_at: The key to end at (i.e. inclusive). An object with this key doesn't have to exist.
972 :param max_workers: Maximum concurrent workers for provider-level recursive listing.
973 :param look_ahead: Prefixes to buffer per worker for provider-level recursive listing.
974 :param include_url_prefix: Whether to include the URL prefix ``msc://profile`` in the result.
975 :param patterns: PatternList for include/exclude filtering. If None, all files are included.
976 :param symlink_handling: How to handle symbolic links during listing. Only applicable for POSIX file storage providers.
977 :return: An iterator over ObjectMetadata for matching files.
978 """
979 pattern_matcher = PatternMatcher(patterns) if patterns else None
980
981 single_file, effective_path = self._resolve_single_file(
982 path, start_after, end_at, include_url_prefix, pattern_matcher
983 )
984 if single_file is not None:
985 yield single_file
986 return
987 if effective_path is None:
988 return
989
990 if self._metadata_provider:
991 objects = self._metadata_provider.list_objects(
992 effective_path,
993 start_after=start_after,
994 end_at=end_at,
995 include_directories=False,
996 )
997 else:
998 objects = self._storage_provider.list_objects_recursive(
999 effective_path,
1000 start_after=start_after,
1001 end_at=end_at,
1002 max_workers=max_workers,
1003 look_ahead=look_ahead,
1004 symlink_handling=symlink_handling,
1005 )
1006
1007 yield from self._filter_and_decorate(objects, include_url_prefix, pattern_matcher)
1008
[docs]
1009 def open(
1010 self,
1011 path: str,
1012 mode: str = "rb",
1013 buffering: int = -1,
1014 encoding: str | None = None,
1015 disable_read_cache: bool = False,
1016 memory_load_limit: int = MEMORY_LOAD_LIMIT,
1017 atomic: bool = True,
1018 check_source_version: SourceVersionCheckMode = SourceVersionCheckMode.INHERIT,
1019 attributes: dict[str, Any] | None = None,
1020 prefetch_file: bool | None = None,
1021 ) -> PosixFile | ObjectFile:
1022 """
1023 Open a file for reading or writing.
1024
1025 :param path: The logical path of the object to open.
1026 :param mode: The file mode. Supported modes: "r", "rb", "w", "wb", "a", "ab".
1027 :param buffering: The buffering mode. Only applies to PosixFile.
1028 :param encoding: The encoding to use for text files.
1029 :param disable_read_cache: When set to ``True``, disables caching for file content.
1030 This parameter is only applicable to ObjectFile when the mode is "r" or "rb".
1031 :param memory_load_limit: Size limit in bytes for loading files into memory. Defaults to 512MB.
1032 This parameter is only applicable to ObjectFile when the mode is "r" or "rb". Defaults to 512MB.
1033 :param atomic: When set to ``True``, file will be written atomically (rename upon close).
1034 This parameter is only applicable to PosixFile in write mode.
1035 :param check_source_version: Whether to check the source version of cached objects.
1036 :param attributes: Attributes to add to the file.
1037 This parameter is only applicable when the mode is "w" or "wb" or "a" or "ab". Defaults to None.
1038 :param prefetch_file: Whether to prefetch the file content.
1039 This parameter is only applicable to ObjectFile when the mode is "r" or "rb".
1040 If None, inherits from cache configuration.
1041 :return: A file-like object (PosixFile or ObjectFile) for the specified path.
1042 :raises FileNotFoundError: If the file does not exist (read mode).
1043 """
1044 if self._is_posix_file_storage_provider():
1045 return PosixFile(
1046 self, path=path, mode=mode, buffering=buffering, encoding=encoding, atomic=atomic, attributes=attributes
1047 )
1048 else:
1049 if atomic is False:
1050 logger.warning("Non-atomic writes are not supported for object storage providers.")
1051
1052 return ObjectFile(
1053 self,
1054 remote_path=path,
1055 mode=mode,
1056 encoding=encoding,
1057 disable_read_cache=disable_read_cache,
1058 memory_load_limit=memory_load_limit,
1059 check_source_version=check_source_version,
1060 attributes=attributes,
1061 prefetch_file=prefetch_file,
1062 )
1063
[docs]
1064 def get_posix_path(self, path: str) -> str | None:
1065 """
1066 Returns the physical POSIX filesystem path for POSIX storage providers.
1067
1068 :param path: The path to resolve (may be a symlink or virtual path).
1069 :return: Physical POSIX filesystem path if POSIX storage, None otherwise.
1070 """
1071 if not self._is_posix_file_storage_provider():
1072 return None
1073
1074 if self._metadata_provider:
1075 resolved = self._metadata_provider.realpath(path)
1076 realpath = resolved.physical_path
1077 else:
1078 realpath = path
1079
1080 return cast(PosixFileStorageProvider, self._storage_provider)._prepend_base_path(realpath)
1081
[docs]
1082 def is_file(self, path: str) -> bool:
1083 """
1084 Checks whether the specified path points to a file (rather than a folder or directory).
1085
1086 :param path: The logical path to check.
1087 :return: ``True`` if the key points to a file, ``False`` otherwise.
1088 """
1089 if self._metadata_provider:
1090 resolved = self._metadata_provider.realpath(path)
1091 return resolved.exists
1092
1093 return self._storage_provider.is_file(path)
1094
1114
[docs]
1115 def is_empty(self, path: str) -> bool:
1116 """
1117 Check whether the specified path is empty. A path is considered empty if there are no
1118 objects whose keys start with the given path as a prefix.
1119
1120 :param path: The logical path to check (typically a directory or folder prefix).
1121 :return: ``True`` if no objects exist under the specified path prefix, ``False`` otherwise.
1122 """
1123 if self._metadata_provider:
1124 objects = self._metadata_provider.list_objects(path)
1125 else:
1126 objects = self._storage_provider.list_objects(path)
1127
1128 try:
1129 return next(objects) is None
1130 except StopIteration:
1131 pass
1132
1133 return True
1134
[docs]
1135 def sync_from(
1136 self,
1137 source_client: AbstractStorageClient,
1138 source_path: str = "",
1139 target_path: str = "",
1140 delete_unmatched_files: bool = False,
1141 description: str = "Syncing",
1142 num_worker_processes: int | None = None,
1143 execution_mode: ExecutionMode = ExecutionMode.LOCAL,
1144 patterns: PatternList | None = None,
1145 preserve_source_attributes: bool = False,
1146 source_files: list[str] | None = None,
1147 ignore_hidden: bool = True,
1148 commit_metadata: bool = True,
1149 dryrun: bool = False,
1150 dryrun_output_path: str | None = None,
1151 symlink_handling: SymlinkHandling = SymlinkHandling.FOLLOW,
1152 ) -> SyncResult:
1153 """
1154 Syncs files from the source storage client to "path/".
1155
1156 :param source_client: The source storage client.
1157 :param source_path: The logical path to sync from.
1158 :param target_path: The logical path to sync to.
1159 :param delete_unmatched_files: Whether to delete files at the target that are not present at the source.
1160 :param description: Description of sync process for logging purposes.
1161 :param num_worker_processes: The number of worker processes to use.
1162 :param execution_mode: The execution mode to use. Currently supports "local" and "ray".
1163 :param patterns: PatternList for include/exclude filtering. If None, all files are included.
1164 Cannot be used together with source_files.
1165 :param preserve_source_attributes: Whether to preserve source file metadata attributes during synchronization.
1166 When ``False`` (default), only file content is copied. When ``True``, custom metadata attributes are also preserved.
1167
1168 .. warning::
1169 **Performance Impact**: When enabled without a ``metadata_provider`` configured, this will make a HEAD
1170 request for each object to retrieve attributes, which can significantly impact performance on large-scale
1171 sync operations. For production use at scale, configure a ``metadata_provider`` in your storage profile.
1172
1173 :param source_files: Optional list of file paths (relative to source_path) to sync. When provided, only these
1174 specific files will be synced, skipping enumeration of the source path. Cannot be used together with patterns.
1175 :param ignore_hidden: Whether to ignore hidden files and directories. Default is ``True``.
1176 :param commit_metadata: When ``True`` (default), calls :py:meth:`StorageClient.commit_metadata` after sync completes.
1177 Set to ``False`` to skip the commit, allowing batching of multiple sync operations before committing manually.
1178 :param dryrun: If ``True``, only enumerate and compare objects without performing any copy/delete operations.
1179 The returned :py:class:`SyncResult` will include a :py:class:`DryrunResult` with paths to JSONL files.
1180 :param dryrun_output_path: Directory to write dryrun JSONL files into. If ``None`` (default), a temporary
1181 directory is created automatically. Ignored when ``dryrun`` is ``False``.
1182 :param symlink_handling: How to handle symbolic links during sync.
1183 :py:attr:`SymlinkHandling.FOLLOW` (default) dereferences symlinks and copies the target's bytes.
1184 :py:attr:`SymlinkHandling.SKIP` excludes symlinks from the sync.
1185 :py:attr:`SymlinkHandling.PRESERVE` recreates symlinks on the target via :py:meth:`make_symlink`
1186 instead of copying bytes (required for round-trip preservation of symlinks).
1187 :raises ValueError: If both source_files and patterns are provided.
1188 :raises RuntimeError: If errors occur during sync operations. The sync will stop on first error (fail-fast).
1189 """
1190 if source_files and patterns:
1191 raise ValueError("Cannot specify both 'source_files' and 'patterns'. Please use only one filtering method.")
1192
1193 pattern_matcher = PatternMatcher(patterns) if patterns else None
1194
1195 # Disable the replica manager during sync
1196 if not isinstance(source_client, NullStorageClient) and source_client._replica_manager:
1197 # Import here to avoid circular dependency
1198 from .client import StorageClient as StorageClientFacade
1199
1200 source_client = StorageClientFacade(source_client._config)
1201 source_client._replica_manager = None
1202
1203 m = SyncManager(source_client, source_path, self, target_path)
1204 batch_size = int(os.environ.get("MSC_SYNC_BATCH_SIZE", DEFAULT_SYNC_BATCH_SIZE))
1205
1206 return m.sync_objects(
1207 execution_mode=execution_mode,
1208 description=description,
1209 num_worker_processes=num_worker_processes,
1210 delete_unmatched_files=delete_unmatched_files,
1211 pattern_matcher=pattern_matcher,
1212 preserve_source_attributes=preserve_source_attributes,
1213 symlink_handling=symlink_handling,
1214 source_files=source_files,
1215 ignore_hidden=ignore_hidden,
1216 commit_metadata=commit_metadata,
1217 batch_size=batch_size,
1218 dryrun=dryrun,
1219 dryrun_output_path=dryrun_output_path,
1220 )
1221
[docs]
1222 def sync_replicas(
1223 self,
1224 source_path: str,
1225 replica_indices: Sequence[int] | None = None,
1226 delete_unmatched_files: bool = False,
1227 description: str = "Syncing replica",
1228 num_worker_processes: int | None = None,
1229 execution_mode: ExecutionMode = ExecutionMode.LOCAL,
1230 patterns: PatternList | None = None,
1231 ignore_hidden: bool = True,
1232 symlink_handling: SymlinkHandling = SymlinkHandling.FOLLOW,
1233 ) -> None:
1234 """
1235 Sync files from this client to its replica storage clients.
1236
1237 :param source_path: The logical path to sync from.
1238 :param replica_indices: Specific replica indices to sync to (0-indexed). If None, syncs to all writable replicas.
1239 Read-only replicas are skipped by default; explicitly selecting one raises ``ValueError``.
1240 :param delete_unmatched_files: When set to ``True``, delete files in replicas that don't exist in source.
1241 :param description: Description of sync process for logging purposes.
1242 :param num_worker_processes: Number of worker processes for parallel sync.
1243 :param execution_mode: Execution mode (LOCAL or REMOTE).
1244 :param patterns: PatternList for include/exclude filtering. If None, all files are included.
1245 :param ignore_hidden: When set to ``True``, ignore hidden files (starting with '.'). Defaults to ``True``.
1246 :param symlink_handling: How to handle symbolic links during sync.
1247 :py:attr:`SymlinkHandling.FOLLOW` (default) dereferences symlinks and copies the target's bytes.
1248 :py:attr:`SymlinkHandling.SKIP` excludes symlinks from the sync.
1249 :py:attr:`SymlinkHandling.PRESERVE` recreates symlinks on each replica via
1250 :py:meth:`make_symlink` instead of copying bytes.
1251 """
1252 if not self._replicas:
1253 logger.warning(
1254 "No replicas found in profile '%s'. Add a 'replicas' section to your profile configuration to enable "
1255 "secondary storage locations for redundancy and performance.",
1256 self._config.profile,
1257 )
1258 return
1259
1260 if replica_indices:
1261 try:
1262 replicas = [self._replicas[i] for i in replica_indices]
1263 except IndexError as e:
1264 raise ValueError(f"Replica index out of range: {replica_indices}") from e
1265 for index, replica in zip(replica_indices, replicas):
1266 if replica not in self._writable_replicas:
1267 raise ValueError(f"Replica index {index} ('{replica.profile}') is read-only and cannot be synced")
1268 else:
1269 replicas = self._writable_replicas
1270
1271 if not replicas:
1272 logger.warning("No writable replicas configured for profile '%s'; sync skipped", self._config.profile)
1273 return
1274
1275 # Disable the replica manager during sync
1276 if self._replica_manager:
1277 # Import here to avoid circular dependency
1278 from .client import StorageClient as StorageClientFacade
1279
1280 source_client = StorageClientFacade(self._config)
1281 source_client._replica_manager = None
1282 else:
1283 source_client = self
1284
1285 for replica in replicas:
1286 replica.sync_from(
1287 source_client,
1288 source_path,
1289 source_path,
1290 delete_unmatched_files=delete_unmatched_files,
1291 description=f"{description} ({replica.profile})",
1292 num_worker_processes=num_worker_processes,
1293 execution_mode=execution_mode,
1294 patterns=patterns,
1295 ignore_hidden=ignore_hidden,
1296 symlink_handling=symlink_handling,
1297 )
1298
[docs]
1299 def list(
1300 self,
1301 path: str = "",
1302 start_after: str | None = None,
1303 end_at: str | None = None,
1304 include_directories: bool = False,
1305 include_url_prefix: bool = False,
1306 attribute_filter_expression: str | None = None,
1307 show_attributes: bool = False,
1308 patterns: PatternList | None = None,
1309 symlink_handling: SymlinkHandling = SymlinkHandling.FOLLOW,
1310 ) -> Iterator[ObjectMetadata]:
1311 """
1312 List objects in the storage provider under the specified path.
1313
1314 :param path: The directory or file path to list objects under. This should be a
1315 complete filesystem path (e.g., "my-bucket/documents/" or "data/2024/").
1316 :param start_after: The key to start after (i.e. exclusive). An object with this key doesn't have to exist.
1317 :param end_at: The key to end at (i.e. inclusive). An object with this key doesn't have to exist.
1318 :param include_directories: Whether to include directories in the result. When ``True``, directories are returned alongside objects.
1319 :param include_url_prefix: Whether to include the URL prefix ``msc://profile`` in the result.
1320 :param attribute_filter_expression: The attribute filter expression to apply to the result.
1321 :param show_attributes: Whether to return attributes in the result. WARNING: Depending on implementation, there may be a performance impact if this is set to ``True``.
1322 :param patterns: PatternList for include/exclude filtering. If None, all files are included.
1323 :param symlink_handling: How to handle symbolic links during listing. Only applicable for POSIX file storage providers.
1324 :return: An iterator over ObjectMetadata for matching objects.
1325 """
1326 pattern_matcher = PatternMatcher(patterns) if patterns else None
1327
1328 single_file, effective_path = self._resolve_single_file(
1329 path, start_after, end_at, include_url_prefix, pattern_matcher
1330 )
1331 if single_file is not None:
1332 evaluator = (
1333 create_attribute_filter_evaluator(attribute_filter_expression) if attribute_filter_expression else None
1334 )
1335 if matches_attribute_filter_expression(single_file, evaluator):
1336 yield single_file
1337 return
1338 if effective_path is None:
1339 return
1340
1341 if self._metadata_provider:
1342 objects = self._metadata_provider.list_objects(
1343 effective_path,
1344 start_after=start_after,
1345 end_at=end_at,
1346 include_directories=include_directories,
1347 attribute_filter_expression=attribute_filter_expression,
1348 show_attributes=show_attributes,
1349 )
1350 else:
1351 objects = self._storage_provider.list_objects(
1352 effective_path,
1353 start_after=start_after,
1354 end_at=end_at,
1355 include_directories=include_directories,
1356 attribute_filter_expression=attribute_filter_expression,
1357 show_attributes=show_attributes,
1358 symlink_handling=symlink_handling,
1359 )
1360
1361 yield from self._filter_and_decorate(objects, include_url_prefix, pattern_matcher)
1362
[docs]
1363 def generate_presigned_url(
1364 self,
1365 path: str,
1366 *,
1367 method: str = "GET",
1368 signer_type: SignerType | None = None,
1369 signer_options: dict[str, Any] | None = None,
1370 ) -> str:
1371 if self._metadata_provider:
1372 if method.upper() not in ("GET", "HEAD"):
1373 raise ValueError(
1374 f"Presigned {method.upper()} URLs are not supported when a metadata provider is configured; "
1375 "uploads through such a URL would bypass the metadata provider."
1376 )
1377 path = self._resolve_read_path(path)
1378 return self._storage_provider.generate_presigned_url(
1379 path, method=method, signer_type=signer_type, signer_options=signer_options
1380 )