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