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