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 codecs
17import logging
18import os
19import stat
20from pathlib import Path, PurePath, PurePosixPath
21from types import NotImplementedType
22
23from .client import StorageClient
24from .shortcuts import resolve_storage_client
25from .types import MSC_PROTOCOL, ObjectMetadata
26from .utils import join_paths
27
28logger = logging.getLogger(__name__)
29
30
[docs]
31class StatResult:
32 """
33 A stat-like result object that mimics os.stat_result for remote storage paths.
34
35 This class provides the same interface as os.stat_result but is populated
36 from ObjectMetadata obtained from storage providers.
37 """
38
39 def __init__(self, metadata: ObjectMetadata):
40 """Initialize StatResult from ObjectMetadata."""
41 # File type and mode bits
42 if metadata.type == "directory":
43 # Directory: 0o755 (rwxr-xr-x) + S_IFDIR
44 self.st_mode = stat.S_IFDIR | 0o755
45 else:
46 # Regular file: 0o644 (rw-r--r--) + S_IFREG
47 self.st_mode = stat.S_IFREG | 0o644
48
49 # File size
50 self.st_size = metadata.content_length
51
52 # Timestamps - convert datetime to epoch seconds
53 mtime = metadata.last_modified.timestamp()
54 self.st_mtime = mtime
55 self.st_atime = mtime
56 self.st_ctime = mtime
57
58 # Nanosecond precision timestamps
59 mtime_ns = int(mtime * 1_000_000_000)
60 self.st_mtime_ns = mtime_ns
61 self.st_atime_ns = mtime_ns
62 self.st_ctime_ns = mtime_ns
63
64 # Default values for fields we don't have from storage providers
65 self.st_ino = 0
66 self.st_dev = 0
67 self.st_nlink = 1
68 self.st_uid = os.getuid() if hasattr(os, "getuid") else 0 # User ID
69 self.st_gid = os.getgid() if hasattr(os, "getgid") else 0 # Group ID
70
71
[docs]
72class MultiStoragePath:
73 """
74 A path object similar to pathlib.Path that supports both local and remote file systems.
75
76 MultiStoragePath provides a unified interface for working with paths across different storage systems,
77 including local files, S3, GCS, Azure Blob Storage, and more. It uses the "msc://" protocol
78 prefix to identify remote storage paths.
79
80 This implementation is based on Python 3.9's pathlib.Path interface, providing compatible behavior
81 for local filesystem operations while extending support to remote storage systems.
82
83 Examples:
84 >>> import multistorageclient as msc
85 >>> msc.Path("/local/path/file.txt")
86 >>> msc.Path("msc://my-profile/data/file.txt")
87 >>> msc.Path(pathlib.Path("relative/path"))
88 """
89
90 _internal_path: PurePosixPath
91 _storage_client: StorageClient
92 _path: str
93
94 def __init__(self, path: str | os.PathLike):
95 """
96 Initialize path object supporting multiple storage backends.
97
98 :param path: String, Path, or MultiStoragePath. Relative paths are automatically converted to absolute.
99 """
100 self._path = str(path)
101 self._storage_client, relative_path = resolve_storage_client(self._path)
102 self._internal_path = PurePosixPath(relative_path)
103
104 if self._storage_client.is_default_profile():
105 self._internal_path = PurePosixPath("/") / self._internal_path
106
107 def __str__(self) -> str:
108 if self._storage_client.is_default_profile():
109 return str(self._internal_path)
110 return join_paths(f"{MSC_PROTOCOL}{self._storage_client.profile}", str(self._internal_path))
111
112 def __repr__(self) -> str:
113 return f"MultiStoragePath({str(self)!r})"
114
115 def __eq__(self, other) -> bool:
116 if not isinstance(other, MultiStoragePath):
117 return False
118 return (
119 self._storage_client.profile == other._storage_client.profile
120 and self._internal_path == other._internal_path
121 )
122
123 def __lt__(self, other):
124 """
125 Return True if this path sorts before another path-like object.
126
127 Ordering is based on the normalized storage profile and internal path. ``pathlib`` paths are
128 compared as local filesystem paths; unsupported types return ``NotImplemented``.
129 """
130 other = self._coerce_path(other)
131 if other is NotImplemented:
132 return NotImplemented
133 return self._ordering_key() < other._ordering_key()
134
135 def __le__(self, other):
136 """Return whether this path sorts before or at another path-like object."""
137 other = self._coerce_path(other)
138 if other is NotImplemented:
139 return NotImplemented
140 return self._ordering_key() <= other._ordering_key()
141
142 def __gt__(self, other):
143 """Return whether this path sorts after another path-like object."""
144 other = self._coerce_path(other)
145 if other is NotImplemented:
146 return NotImplemented
147 return self._ordering_key() > other._ordering_key()
148
149 def __ge__(self, other):
150 """Return whether this path sorts after or at another path-like object."""
151 other = self._coerce_path(other)
152 if other is NotImplemented:
153 return NotImplemented
154 return self._ordering_key() >= other._ordering_key()
155
156 @staticmethod
157 def _coerce_path(other) -> "MultiStoragePath | NotImplementedType":
158 """
159 Convert supported path-like objects to ``MultiStoragePath`` for comparison.
160 """
161 if isinstance(other, MultiStoragePath):
162 return other
163 if isinstance(other, PurePath):
164 return MultiStoragePath(other)
165 return NotImplemented
166
167 def _ordering_key(self) -> tuple[str, PurePosixPath]:
168 """
169 Return the resolved profile and internal path used for equality-compatible ordering.
170 """
171 return (self._storage_client.profile, self._internal_path)
172
173 def __hash__(self) -> int:
174 """Return hash of the path."""
175 return hash((self._storage_client.profile, self._internal_path))
176
177 def __fspath__(self) -> str:
178 return str(self)
179
[docs]
180 def joinpath(self, *pathsegments):
181 return self.with_segments(*pathsegments)
182
183 def __truediv__(self, key):
184 try:
185 return self.joinpath(key)
186 except TypeError:
187 return NotImplemented
188
189 def __rtruediv__(self, key):
190 try:
191 return self.with_segments(key, self)
192 except TypeError:
193 return NotImplemented
194
195 def __getstate__(self):
196 return {"_path": self._path, "_internal_path": self._internal_path}
197
198 def __setstate__(self, state):
199 self._path = state["_path"]
200 self._internal_path = state["_internal_path"]
201 self._storage_client, _ = resolve_storage_client(self._path)
202
203 @property
204 def anchor(self) -> str:
205 """
206 The concatenation of the drive and root, or ''.
207 """
208 return self._internal_path.anchor
209
210 @property
211 def name(self) -> str:
212 """
213 The final path component, if any.
214 """
215 return self._internal_path.name
216
217 @property
218 def suffix(self) -> str:
219 """
220 The final path component, if any.
221 """
222 return self._internal_path.suffix
223
224 @property
225 def suffixes(self) -> list[str]:
226 """
227 A list of the final component's suffixes, if any.
228
229 These include the leading periods. For example: ['.tar', '.gz']
230 """
231 return self._internal_path.suffixes
232
233 @property
234 def stem(self) -> str:
235 """
236 The final path component, minus its last suffix.
237 """
238 return self._internal_path.stem
239
240 @property
241 def parent(self) -> "MultiStoragePath":
242 """
243 The logical parent of the path.
244 """
245 parent_path = self._internal_path.parent
246 if self._storage_client.is_default_profile():
247 return MultiStoragePath(str(parent_path))
248 return MultiStoragePath(join_paths(f"{MSC_PROTOCOL}{self._storage_client.profile}", str(parent_path)))
249
250 @property
251 def parents(self) -> list["MultiStoragePath"]:
252 """
253 A sequence of this path's logical parents.
254 """
255 if self._storage_client.is_default_profile():
256 return [MultiStoragePath(str(p)) for p in self._internal_path.parents]
257 else:
258 return [
259 MultiStoragePath(join_paths(f"{MSC_PROTOCOL}{self._storage_client.profile}", str(p)))
260 for p in self._internal_path.parents
261 ]
262
263 @property
264 def parts(self):
265 """
266 An object providing sequence-like access to the components in the filesystem path (does not
267 include the msc:// and the profile name).
268 """
269 return self._internal_path.parts
270
[docs]
271 def as_posix(self) -> str:
272 """
273 Return the string representation of the path with forward (/) slashes.
274
275 If the path is a remote path, the file content is downloaded to local storage
276 (either cached or temporary file) and the local filesystem path is returned.
277 This enables access to remote file content through standard filesystem operations.
278 """
279 if self._storage_client.is_default_profile():
280 return self._internal_path.as_posix()
281
282 # Return the local path of the file
283 with self._storage_client.open(str(self._internal_path), mode="rb") as fp:
284 return fp.resolve_filesystem_path()
285
[docs]
286 def is_absolute(self) -> bool:
287 """
288 Paths are always absolute.
289 """
290 return True
291
[docs]
292 def is_relative_to(self, other: "MultiStoragePath") -> bool:
293 """
294 Return True if the path is relative to another path in the same storage profile, or False otherwise.
295 """
296 return (
297 isinstance(other, MultiStoragePath)
298 and self._storage_client.profile == other._storage_client.profile
299 and self._internal_path.is_relative_to(other._internal_path)
300 )
301
[docs]
302 def is_reserved(self) -> bool:
303 if self._storage_client.is_default_profile():
304 return self._internal_path.is_reserved()
305 raise NotImplementedError("MultiStoragePath.is_reserved() is unsupported for remote storage paths")
306
[docs]
307 def match(self, pattern) -> bool:
308 """
309 Return True if this path matches the given pattern.
310 """
311 return Path(self._internal_path).match(pattern)
312
[docs]
313 def relative_to(self, other: "MultiStoragePath") -> PurePosixPath:
314 """
315 Return a version of this path relative to another path.
316
317 Both paths must use the same storage profile. The operation raises ValueError if:
318 - The paths have different storage profiles
319 - This path is not relative to the other path
320 - The other path is not a MultiStoragePath instance
321
322 Note: This method returns a PurePosixPath (not a MultiStoragePath) because
323 MultiStoragePath always represents absolute paths.
324
325 :param other: The base path to calculate relative path from
326 :return: A PurePosixPath representing the relative path
327 :raises ValueError: If the path cannot be made relative to other
328 :raises TypeError: If other is not a MultiStoragePath instance
329 """
330 if not isinstance(other, MultiStoragePath):
331 raise TypeError(f"'{type(other).__name__}' object cannot be used as relative base")
332
333 # Check if both paths use the same storage profile
334 if self._storage_client.profile != other._storage_client.profile:
335 raise ValueError(
336 f"Cannot compute relative path between different storage profiles: "
337 f"'{self._storage_client.profile}' and '{other._storage_client.profile}'"
338 )
339
340 # Use the internal PurePosixPath.relative_to() method
341 try:
342 return self._internal_path.relative_to(other._internal_path)
343 except ValueError:
344 # Re-raise with paths that include the profile information for clarity
345 raise ValueError(f"{str(self)!r} is not in the subpath of {str(other)!r}")
346
[docs]
347 def with_name(self, name: str) -> "MultiStoragePath":
348 """
349 Return a new path with the file name changed.
350 """
351 if self._storage_client.is_default_profile():
352 return MultiStoragePath(str(self._internal_path.with_name(name)))
353 else:
354 return MultiStoragePath(
355 join_paths(f"{MSC_PROTOCOL}{self._storage_client.profile}", str(self._internal_path.with_name(name)))
356 )
357
[docs]
358 def with_stem(self, stem: str) -> "MultiStoragePath":
359 """
360 Return a new path with the stem changed.
361 """
362 if self._storage_client.is_default_profile():
363 return MultiStoragePath(str(self._internal_path.with_stem(stem)))
364 else:
365 return MultiStoragePath(
366 join_paths(f"{MSC_PROTOCOL}{self._storage_client.profile}", str(self._internal_path.with_stem(stem)))
367 )
368
[docs]
369 def with_suffix(self, suffix: str) -> "MultiStoragePath":
370 """
371 Return a new path with the file suffix changed. If the path has no suffix, add given suffix.
372 If the given suffix is an empty string, remove the suffix from the path.
373 """
374 if self._storage_client.is_default_profile():
375 return MultiStoragePath(str(self._internal_path.with_suffix(suffix)))
376 else:
377 return MultiStoragePath(
378 join_paths(
379 f"{MSC_PROTOCOL}{self._storage_client.profile}", str(self._internal_path.with_suffix(suffix))
380 )
381 )
382
[docs]
383 def with_segments(self, *pathsegments) -> "MultiStoragePath":
384 """
385 Construct a new path object from any number of path-like objects.
386 """
387 if self._storage_client.is_default_profile():
388 new_path = self._internal_path.joinpath(*pathsegments)
389 return MultiStoragePath(str(new_path))
390 else:
391 new_path = self._internal_path.joinpath(*pathsegments)
392 return MultiStoragePath(join_paths(f"{MSC_PROTOCOL}{self._storage_client.profile}", str(new_path)))
393
394 # Expanding and resolving paths
395
[docs]
396 @classmethod
397 def home(cls):
398 """
399 Return a new path pointing to the user's home directory.
400 """
401 return Path.home()
402
[docs]
403 def expanduser(self):
404 """
405 Return a new path with expanded ~ and ~user constructs (as returned by os.path.expanduser).
406
407 Not supported for remote storage paths.
408 """
409 if self._storage_client.is_default_profile():
410 return Path(self._internal_path).expanduser()
411 raise NotImplementedError("MultiStoragePath.expanduser() is unsupported for remote storage paths")
412
[docs]
413 @classmethod
414 def cwd(cls):
415 """
416 Return a new path pointing to the current working directory.
417 """
418 return Path.cwd()
419
[docs]
420 def absolute(self):
421 """
422 Return the path itself since it is always absolute.
423 """
424 return self
425
[docs]
426 def resolve(self, strict=False):
427 """
428 Return the absolute path.
429 """
430 if self._storage_client.is_default_profile():
431 return MultiStoragePath(str(Path(self._internal_path).resolve(strict=strict)))
432 return MultiStoragePath(join_paths(f"{MSC_PROTOCOL}{self._storage_client.profile}", str(self._internal_path)))
433
[docs]
434 def readlink(self):
435 """
436 Return the path to which the symbolic link points.
437
438 Not supported for remote storage paths.
439 """
440 if self._storage_client.is_default_profile():
441 return MultiStoragePath(str(Path(self._internal_path).readlink()))
442 raise NotImplementedError("MultiStoragePath.readlink() is unsupported for remote storage paths")
443
444 # Querying file type and status
445
[docs]
446 def stat(self):
447 """
448 Return the result of the stat() system call on this path, like os.stat() does.
449
450 If the path is a remote path, the result is a :py:class:`multistorageclient.pathlib.StatResult` object.
451 """
452 if self._storage_client.is_default_profile():
453 return Path(self._internal_path).stat()
454 info = self._storage_client.info(str(self._internal_path))
455 return StatResult(info)
456
[docs]
457 def lstat(self):
458 """
459 Like stat(), except if the path points to a symlink, the symlink's status information
460 is returned, rather than its target's.
461
462 If the path is a remote path, the result is a :py:class:`multistorageclient.pathlib.StatResult` object.
463 """
464 if self._storage_client.is_default_profile():
465 return Path(self._internal_path).lstat()
466 info = self._storage_client.info(str(self._internal_path))
467 return StatResult(info)
468
[docs]
469 def exists(self) -> bool:
470 """
471 Return True if the path exists.
472 """
473 if self._storage_client.is_default_profile():
474 return Path(self._internal_path).exists()
475 else:
476 try:
477 self._storage_client.info(str(self._internal_path))
478 return True
479 except FileNotFoundError:
480 return False
481
[docs]
482 def is_file(self, strict: bool = True) -> bool:
483 """
484 Return True if the path exists and is a regular file.
485 """
486 if self._storage_client.is_default_profile():
487 return Path(self._internal_path).is_file()
488 else:
489 try:
490 # If the path ends with a "/", assume it is a directory.
491 path = str(self._internal_path)
492 if path.endswith("/"):
493 return False
494
495 meta = self._storage_client.info(path, strict=strict)
496 return meta.type == "file"
497 except FileNotFoundError:
498 return False
499 except Exception as e:
500 logger.warning("Error occurred while fetching file info at %s, caused by: %s", self._internal_path, e)
501 return False
502
[docs]
503 def is_dir(self, strict: bool = True) -> bool:
504 """
505 Return True if the path exists and is a directory.
506 """
507 if self._storage_client.is_default_profile():
508 return Path(self._internal_path).is_dir()
509 else:
510 try:
511 # If the path does not end with a "/", append it to ensure the path is a directory.
512 path = str(self._internal_path)
513 if not path.endswith("/"):
514 path += "/"
515
516 meta = self._storage_client.info(path, strict=strict)
517 return meta.type == "directory"
518 except FileNotFoundError:
519 return False
520 except Exception as e:
521 logger.warning("Error occurred while fetching file info at %s, caused by: %s", self._internal_path, e)
522 return False
523
[docs]
524 def is_symlink(self):
525 """
526 Return True if the path exists and is a symbolic link.
527
528 Not supported for remote storage paths.
529 """
530 if self._storage_client.is_default_profile():
531 return Path(self._internal_path).is_symlink()
532 raise NotImplementedError("MultiStoragePath.is_symlink() is unsupported for remote storage paths")
533
[docs]
534 def is_mount(self):
535 """
536 Return True if the path exists and is a mount point.
537
538 Not supported for remote storage paths.
539 """
540 if self._storage_client.is_default_profile():
541 return Path(self._internal_path).is_mount()
542 raise NotImplementedError("MultiStoragePath.is_mount() is unsupported for remote storage paths")
543
[docs]
544 def is_socket(self):
545 """
546 Return True if the path exists and is a socket.
547
548 Not supported for remote storage paths.
549 """
550 if self._storage_client.is_default_profile():
551 return Path(self._internal_path).is_socket()
552 raise NotImplementedError("MultiStoragePath.is_socket() is unsupported for remote storage paths")
553
[docs]
554 def is_fifo(self):
555 """
556 Return True if the path exists and is a FIFO.
557
558 Not supported for remote storage paths.
559 """
560 if self._storage_client.is_default_profile():
561 return Path(self._internal_path).is_fifo()
562 raise NotImplementedError("MultiStoragePath.is_fifo() is unsupported for remote storage paths")
563
[docs]
564 def is_block_device(self):
565 """
566 Return True if the path exists and is a block device.
567
568 Not supported for remote storage paths.
569 """
570 if self._storage_client.is_default_profile():
571 return Path(self._internal_path).is_block_device()
572 raise NotImplementedError("MultiStoragePath.is_block_device() is unsupported for remote storage paths")
573
[docs]
574 def is_char_device(self):
575 """
576 Return True if the path exists and is a character device.
577
578 Not supported for remote storage paths.
579 """
580 if self._storage_client.is_default_profile():
581 return Path(self._internal_path).is_char_device()
582 raise NotImplementedError("MultiStoragePath.is_char_device() is unsupported for remote storage paths")
583
[docs]
584 def samefile(self, other_path):
585 """
586 Return True if both paths point to the same file or directory.
587
588 Not supported for remote storage paths.
589 """
590 if self._storage_client.is_default_profile():
591 return Path(self._internal_path).samefile(other_path)
592 return self == other_path
593
594 # Reading and writing files
595
[docs]
596 def open(self, mode="r", buffering=-1, encoding=None, errors=None, newline=None, **kwargs):
597 """
598 Open the file and return a file object.
599
600 :param mode: The file mode to open the file in.
601 :param buffering: The buffering mode.
602 :param encoding: The encoding to use for text files.
603 :param errors: How to handle encoding errors.
604 :param newline: Controls universal newlines mode.
605 :param kwargs: Additional arguments passed to client.open (e.g., check_source_version, prefetch_file, etc.)
606 """
607 return self._storage_client.open(
608 str(self._internal_path), mode=mode, buffering=buffering, encoding=encoding, **kwargs
609 )
610
[docs]
611 def read_bytes(self) -> bytes:
612 """
613 Open the file in bytes mode, read it, and close the file.
614 """
615 return self._storage_client.read(str(self._internal_path))
616
[docs]
617 def read_text(self, encoding: str = "utf-8", errors: str = "strict") -> str:
618 """
619 Open the file in text mode, read it, and close the file.
620 """
621 result = self._storage_client.read(str(self._internal_path))
622 if not hasattr(result, "decode"): # Rust's PyBytes does not implement decode
623 return codecs.decode(memoryview(result), encoding, errors)
624 return result.decode(encoding, errors)
625
[docs]
626 def write_bytes(self, data: bytes) -> None:
627 """
628 Open the file in bytes mode, write to it, and close the file.
629 """
630 self._storage_client.write(str(self._internal_path), data)
631
[docs]
632 def write_text(self, data: str, encoding: str = "utf-8", errors: str = "strict") -> None:
633 """
634 Open the file in text mode, write to it, and close the file.
635 """
636 self._storage_client.write(str(self._internal_path), data.encode(encoding, errors))
637
638 # Reading directories
639
[docs]
640 def iterdir(self):
641 """
642 Yield path objects of the directory contents.
643 """
644 if self._storage_client.is_default_profile():
645 for item in Path(self._internal_path).iterdir():
646 yield MultiStoragePath(str(item))
647 else:
648 path = str(self._internal_path)
649 if not path.endswith("/"):
650 path += "/"
651 for item in self._storage_client.list(path, include_directories=True, include_url_prefix=True):
652 yield MultiStoragePath(item.key)
653
[docs]
654 def glob(self, pattern):
655 """
656 Iterate over this subtree and yield all existing files (of any kind, including directories)
657 matching the given relative pattern.
658 """
659 if self._storage_client.is_default_profile():
660 return [MultiStoragePath(str(p)) for p in Path(self._internal_path).glob(pattern)]
661 else:
662 return [
663 MultiStoragePath(str(p))
664 for p in self._storage_client.glob(str(self._internal_path / pattern), include_url_prefix=True)
665 ]
666
[docs]
667 def rglob(self, pattern):
668 """
669 Recursively yield all existing files (of any kind, including directories) matching the
670 given relative pattern, anywhere in this subtree.
671 """
672 if self._storage_client.is_default_profile():
673 return [MultiStoragePath(str(p)) for p in Path(self._internal_path).rglob(pattern)]
674 else:
675 recursive_pattern = f"**/{pattern}"
676 return [
677 MultiStoragePath(str(p))
678 for p in self._storage_client.glob(
679 str(self._internal_path / recursive_pattern), include_url_prefix=True
680 )
681 ]
682
[docs]
683 def walk(self, top_down=True, on_error=None, follow_symlinks=False):
684 """
685 Walk the directory tree from this directory, similar to os.walk().
686
687 Not supported for remote storage paths.
688 """
689 if self._storage_client.is_default_profile():
690 return Path(self._internal_path).walk(top_down, on_error, follow_symlinks) # pyright: ignore[reportAttributeAccessIssue]
691 raise NotImplementedError("MultiStoragePath.walk() is unsupported for remote storage paths")
692
693 # Creating files and directories
694
[docs]
695 def touch(self, mode=0o666, exist_ok=False):
696 """
697 Create this file with the given access mode, if it doesn't exist.
698 """
699 if self._storage_client.is_default_profile():
700 Path(self._internal_path).touch(mode, exist_ok)
701 else:
702 if self.exists():
703 # object storage does not support updating the last modified time of an object without writing the object
704 logger.warning("MultiStoragePath.touch() is not supported for remote storage paths")
705 else:
706 self._storage_client.write(str(self._internal_path), b"")
707
[docs]
708 def mkdir(self, mode=0o777, parents=False, exist_ok=False) -> None:
709 """
710 Create a new directory at the given path.
711
712 For remote storage paths, this operation is a no-op.
713 """
714 if self._storage_client.is_default_profile():
715 Path(self._internal_path).mkdir(mode, parents, exist_ok)
716
[docs]
717 def symlink_to(self, target, target_is_directory=False):
718 """
719 Make this path a symlink pointing to the target path.
720
721 Not supported for remote storage paths.
722 """
723 if self._storage_client.is_default_profile():
724 Path(self._internal_path).symlink_to(target, target_is_directory)
725 else:
726 raise NotImplementedError("MultiStoragePath.symlink_to() is unsupported for remote storage paths")
727
728 # Renaming and deleting
729
[docs]
730 def rename(self, target) -> "MultiStoragePath":
731 """
732 Rename this path to the target path.
733 """
734 if not isinstance(target, MultiStoragePath):
735 target = MultiStoragePath(target)
736
737 if self._storage_client.is_default_profile():
738 Path(self._internal_path).rename(str(target._internal_path))
739 else:
740 # Note: This operation is not atomic, and the target path must be a single file.
741 self._storage_client.copy(str(self._internal_path), str(target._internal_path))
742 self._storage_client.delete(str(self._internal_path))
743
744 return target
745
[docs]
746 def replace(self, target):
747 """
748 Rename this path to the target path, overwriting if that path exists.
749
750 Not supported for remote storage paths.
751 """
752 if self._storage_client.is_default_profile():
753 Path(self._internal_path).replace(target)
754 else:
755 raise NotImplementedError("MultiStoragePath.replace() is unsupported for remote storage paths")
756
[docs]
757 def unlink(self, missing_ok: bool = False) -> None:
758 """
759 Remove this file or link. If the path is a directory, use rmdir() instead.
760 """
761 if self._storage_client.is_default_profile():
762 Path(self._internal_path).unlink(missing_ok=missing_ok)
763 else:
764 try:
765 self._storage_client.delete(str(self._internal_path))
766 except FileNotFoundError:
767 if not missing_ok:
768 raise
769
[docs]
770 def rmdir(self) -> None:
771 """
772 Remove this directory. The directory must be empty.
773
774 Not supported for remote storage paths.
775 """
776 if self._storage_client.is_default_profile():
777 Path(self._internal_path).rmdir()
778 else:
779 raise NotImplementedError("MultiStoragePath.rmdir() is unsupported for remote storage paths")
780
781 # Permissions and ownership
782
[docs]
783 def owner(self):
784 """
785 Return the login name of the file owner.
786
787 Not supported for remote storage paths.
788 """
789 if self._storage_client.is_default_profile():
790 return Path(self._internal_path).owner()
791 raise NotImplementedError("MultiStoragePath.owner() is unsupported for remote storage paths")
792
[docs]
793 def group(self):
794 """
795 Return the group name of the file gid.
796
797 Not supported for remote storage paths.
798 """
799 if self._storage_client.is_default_profile():
800 return Path(self._internal_path).group()
801 raise NotImplementedError("MultiStoragePath.group() is unsupported for remote storage paths")
802
[docs]
803 def chmod(self, mode):
804 """
805 Change the permissions of the path, like os.chmod().
806
807 Not supported for remote storage paths.
808 """
809 if self._storage_client.is_default_profile():
810 Path(self._internal_path).chmod(mode)
811 else:
812 raise NotImplementedError("MultiStoragePath.chmod() is unsupported for remote storage paths")
813
[docs]
814 def lchmod(self, mode):
815 """
816 Like chmod(), except if the path points to a symlink, the symlink's permissions are changed, rather
817 than its target's.
818
819 Not supported for remote storage paths.
820 """
821 if self._storage_client.is_default_profile():
822 Path(self._internal_path).lchmod(mode)
823 else:
824 raise NotImplementedError("MultiStoragePath.lchmod() is unsupported for remote storage paths")