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