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