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 in the same storage profile, or False otherwise.
276 """
277 return (
278 isinstance(other, MultiStoragePath)
279 and self._storage_client.profile == other._storage_client.profile
280 and self._internal_path.is_relative_to(other._internal_path)
281 )
282
[docs]
283 def is_reserved(self) -> bool:
284 if self._storage_client.is_default_profile():
285 return self._internal_path.is_reserved()
286 raise NotImplementedError("MultiStoragePath.is_reserved() is unsupported for remote storage paths")
287
[docs]
288 def match(self, pattern) -> bool:
289 """
290 Return True if this path matches the given pattern.
291 """
292 return Path(self._internal_path).match(pattern)
293
[docs]
294 def relative_to(self, other: "MultiStoragePath") -> PurePosixPath:
295 """
296 Return a version of this path relative to another path.
297
298 Both paths must use the same storage profile. The operation raises ValueError if:
299 - The paths have different storage profiles
300 - This path is not relative to the other path
301 - The other path is not a MultiStoragePath instance
302
303 Note: This method returns a PurePosixPath (not a MultiStoragePath) because
304 MultiStoragePath always represents absolute paths.
305
306 :param other: The base path to calculate relative path from
307 :return: A PurePosixPath representing the relative path
308 :raises ValueError: If the path cannot be made relative to other
309 :raises TypeError: If other is not a MultiStoragePath instance
310 """
311 if not isinstance(other, MultiStoragePath):
312 raise TypeError(f"'{type(other).__name__}' object cannot be used as relative base")
313
314 # Check if both paths use the same storage profile
315 if self._storage_client.profile != other._storage_client.profile:
316 raise ValueError(
317 f"Cannot compute relative path between different storage profiles: "
318 f"'{self._storage_client.profile}' and '{other._storage_client.profile}'"
319 )
320
321 # Use the internal PurePosixPath.relative_to() method
322 try:
323 return self._internal_path.relative_to(other._internal_path)
324 except ValueError:
325 # Re-raise with paths that include the profile information for clarity
326 raise ValueError(f"{str(self)!r} is not in the subpath of {str(other)!r}")
327
[docs]
328 def with_name(self, name: str) -> "MultiStoragePath":
329 """
330 Return a new path with the file name changed.
331 """
332 if self._storage_client.is_default_profile():
333 return MultiStoragePath(str(self._internal_path.with_name(name)))
334 else:
335 return MultiStoragePath(
336 join_paths(f"{MSC_PROTOCOL}{self._storage_client.profile}", str(self._internal_path.with_name(name)))
337 )
338
[docs]
339 def with_stem(self, stem: str) -> "MultiStoragePath":
340 """
341 Return a new path with the stem changed.
342 """
343 if self._storage_client.is_default_profile():
344 return MultiStoragePath(str(self._internal_path.with_stem(stem)))
345 else:
346 return MultiStoragePath(
347 join_paths(f"{MSC_PROTOCOL}{self._storage_client.profile}", str(self._internal_path.with_stem(stem)))
348 )
349
[docs]
350 def with_suffix(self, suffix: str) -> "MultiStoragePath":
351 """
352 Return a new path with the file suffix changed. If the path has no suffix, add given suffix.
353 If the given suffix is an empty string, remove the suffix from the path.
354 """
355 if self._storage_client.is_default_profile():
356 return MultiStoragePath(str(self._internal_path.with_suffix(suffix)))
357 else:
358 return MultiStoragePath(
359 join_paths(
360 f"{MSC_PROTOCOL}{self._storage_client.profile}", str(self._internal_path.with_suffix(suffix))
361 )
362 )
363
[docs]
364 def with_segments(self, *pathsegments) -> "MultiStoragePath":
365 """
366 Construct a new path object from any number of path-like objects.
367 """
368 if self._storage_client.is_default_profile():
369 new_path = self._internal_path.joinpath(*pathsegments)
370 return MultiStoragePath(str(new_path))
371 else:
372 new_path = self._internal_path.joinpath(*pathsegments)
373 return MultiStoragePath(join_paths(f"{MSC_PROTOCOL}{self._storage_client.profile}", str(new_path)))
374
375 # Expanding and resolving paths
376
[docs]
377 @classmethod
378 def home(cls):
379 """
380 Return a new path pointing to the user's home directory.
381 """
382 return Path.home()
383
[docs]
384 def expanduser(self):
385 """
386 Return a new path with expanded ~ and ~user constructs (as returned by os.path.expanduser).
387
388 Not supported for remote storage paths.
389 """
390 if self._storage_client.is_default_profile():
391 return Path(self._internal_path).expanduser()
392 raise NotImplementedError("MultiStoragePath.expanduser() is unsupported for remote storage paths")
393
[docs]
394 @classmethod
395 def cwd(cls):
396 """
397 Return a new path pointing to the current working directory.
398 """
399 return Path.cwd()
400
[docs]
401 def absolute(self):
402 """
403 Return the path itself since it is always absolute.
404 """
405 return self
406
[docs]
407 def resolve(self, strict=False):
408 """
409 Return the absolute path.
410 """
411 if self._storage_client.is_default_profile():
412 return MultiStoragePath(str(Path(self._internal_path).resolve(strict=strict)))
413 return MultiStoragePath(join_paths(f"{MSC_PROTOCOL}{self._storage_client.profile}", str(self._internal_path)))
414
[docs]
415 def readlink(self):
416 """
417 Return the path to which the symbolic link points.
418
419 Not supported for remote storage paths.
420 """
421 if self._storage_client.is_default_profile():
422 return MultiStoragePath(str(Path(self._internal_path).readlink()))
423 raise NotImplementedError("MultiStoragePath.readlink() is unsupported for remote storage paths")
424
425 # Querying file type and status
426
[docs]
427 def stat(self):
428 """
429 Return the result of the stat() system call on this path, like os.stat() does.
430
431 If the path is a remote path, the result is a :py:class:`multistorageclient.pathlib.StatResult` object.
432 """
433 if self._storage_client.is_default_profile():
434 return Path(self._internal_path).stat()
435 info = self._storage_client.info(str(self._internal_path))
436 return StatResult(info)
437
[docs]
438 def lstat(self):
439 """
440 Like stat(), except if the path points to a symlink, the symlink's status information
441 is returned, rather than its target's.
442
443 If the path is a remote path, the result is a :py:class:`multistorageclient.pathlib.StatResult` object.
444 """
445 if self._storage_client.is_default_profile():
446 return Path(self._internal_path).lstat()
447 info = self._storage_client.info(str(self._internal_path))
448 return StatResult(info)
449
[docs]
450 def exists(self) -> bool:
451 """
452 Return True if the path exists.
453 """
454 if self._storage_client.is_default_profile():
455 return Path(self._internal_path).exists()
456 else:
457 try:
458 self._storage_client.info(str(self._internal_path))
459 return True
460 except FileNotFoundError:
461 return False
462
[docs]
463 def is_file(self, strict: bool = True) -> bool:
464 """
465 Return True if the path exists and is a regular file.
466 """
467 if self._storage_client.is_default_profile():
468 return Path(self._internal_path).is_file()
469 else:
470 try:
471 # If the path ends with a "/", assume it is a directory.
472 path = str(self._internal_path)
473 if path.endswith("/"):
474 return False
475
476 meta = self._storage_client.info(path, strict=strict)
477 return meta.type == "file"
478 except FileNotFoundError:
479 return False
480 except Exception as e:
481 logger.warning("Error occurred while fetching file info at %s, caused by: %s", self._internal_path, e)
482 return False
483
[docs]
484 def is_dir(self, strict: bool = True) -> bool:
485 """
486 Return True if the path exists and is a directory.
487 """
488 if self._storage_client.is_default_profile():
489 return Path(self._internal_path).is_dir()
490 else:
491 try:
492 # If the path does not end with a "/", append it to ensure the path is a directory.
493 path = str(self._internal_path)
494 if not path.endswith("/"):
495 path += "/"
496
497 meta = self._storage_client.info(path, strict=strict)
498 return meta.type == "directory"
499 except FileNotFoundError:
500 return False
501 except Exception as e:
502 logger.warning("Error occurred while fetching file info at %s, caused by: %s", self._internal_path, e)
503 return False
504
[docs]
505 def is_symlink(self):
506 """
507 Return True if the path exists and is a symbolic link.
508
509 Not supported for remote storage paths.
510 """
511 if self._storage_client.is_default_profile():
512 return Path(self._internal_path).is_symlink()
513 raise NotImplementedError("MultiStoragePath.is_symlink() is unsupported for remote storage paths")
514
[docs]
515 def is_mount(self):
516 """
517 Return True if the path exists and is a mount point.
518
519 Not supported for remote storage paths.
520 """
521 if self._storage_client.is_default_profile():
522 return Path(self._internal_path).is_mount()
523 raise NotImplementedError("MultiStoragePath.is_mount() is unsupported for remote storage paths")
524
[docs]
525 def is_socket(self):
526 """
527 Return True if the path exists and is a socket.
528
529 Not supported for remote storage paths.
530 """
531 if self._storage_client.is_default_profile():
532 return Path(self._internal_path).is_socket()
533 raise NotImplementedError("MultiStoragePath.is_socket() is unsupported for remote storage paths")
534
[docs]
535 def is_fifo(self):
536 """
537 Return True if the path exists and is a FIFO.
538
539 Not supported for remote storage paths.
540 """
541 if self._storage_client.is_default_profile():
542 return Path(self._internal_path).is_fifo()
543 raise NotImplementedError("MultiStoragePath.is_fifo() is unsupported for remote storage paths")
544
[docs]
545 def is_block_device(self):
546 """
547 Return True if the path exists and is a block device.
548
549 Not supported for remote storage paths.
550 """
551 if self._storage_client.is_default_profile():
552 return Path(self._internal_path).is_block_device()
553 raise NotImplementedError("MultiStoragePath.is_block_device() is unsupported for remote storage paths")
554
[docs]
555 def is_char_device(self):
556 """
557 Return True if the path exists and is a character device.
558
559 Not supported for remote storage paths.
560 """
561 if self._storage_client.is_default_profile():
562 return Path(self._internal_path).is_char_device()
563 raise NotImplementedError("MultiStoragePath.is_char_device() is unsupported for remote storage paths")
564
[docs]
565 def samefile(self, other_path):
566 """
567 Return True if both paths point to the same file or directory.
568
569 Not supported for remote storage paths.
570 """
571 if self._storage_client.is_default_profile():
572 return Path(self._internal_path).samefile(other_path)
573 return self == other_path
574
575 # Reading and writing files
576
[docs]
577 def open(self, mode="r", buffering=-1, encoding=None, errors=None, newline=None, **kwargs):
578 """
579 Open the file and return a file object.
580
581 :param mode: The file mode to open the file in.
582 :param buffering: The buffering mode.
583 :param encoding: The encoding to use for text files.
584 :param errors: How to handle encoding errors.
585 :param newline: Controls universal newlines mode.
586 :param kwargs: Additional arguments passed to client.open (e.g., check_source_version, prefetch_file, etc.)
587 """
588 return self._storage_client.open(
589 str(self._internal_path), mode=mode, buffering=buffering, encoding=encoding, **kwargs
590 )
591
[docs]
592 def read_bytes(self) -> bytes:
593 """
594 Open the file in bytes mode, read it, and close the file.
595 """
596 return self._storage_client.read(str(self._internal_path))
597
[docs]
598 def read_text(self, encoding: str = "utf-8", errors: str = "strict") -> str:
599 """
600 Open the file in text mode, read it, and close the file.
601 """
602 result = self._storage_client.read(str(self._internal_path))
603 if not hasattr(result, "decode"): # Rust's PyBytes does not implement decode
604 return codecs.decode(memoryview(result), encoding, errors)
605 return result.decode(encoding, errors)
606
[docs]
607 def write_bytes(self, data: bytes) -> None:
608 """
609 Open the file in bytes mode, write to it, and close the file.
610 """
611 self._storage_client.write(str(self._internal_path), data)
612
[docs]
613 def write_text(self, data: str, encoding: str = "utf-8", errors: str = "strict") -> None:
614 """
615 Open the file in text mode, write to it, and close the file.
616 """
617 self._storage_client.write(str(self._internal_path), data.encode(encoding))
618
619 # Reading directories
620
[docs]
621 def iterdir(self):
622 """
623 Yield path objects of the directory contents.
624 """
625 if self._storage_client.is_default_profile():
626 for item in Path(self._internal_path).iterdir():
627 yield MultiStoragePath(str(item))
628 else:
629 path = str(self._internal_path)
630 if not path.endswith("/"):
631 path += "/"
632 for item in self._storage_client.list(path, include_directories=True, include_url_prefix=True):
633 yield MultiStoragePath(item.key)
634
[docs]
635 def glob(self, pattern):
636 """
637 Iterate over this subtree and yield all existing files (of any kind, including directories)
638 matching the given relative pattern.
639 """
640 if self._storage_client.is_default_profile():
641 return [MultiStoragePath(str(p)) for p in Path(self._internal_path).glob(pattern)]
642 else:
643 return [
644 MultiStoragePath(str(p))
645 for p in self._storage_client.glob(str(self._internal_path / pattern), include_url_prefix=True)
646 ]
647
[docs]
648 def rglob(self, pattern):
649 """
650 Recursively yield all existing files (of any kind, including directories) matching the
651 given relative pattern, anywhere in this subtree.
652 """
653 if self._storage_client.is_default_profile():
654 return [MultiStoragePath(str(p)) for p in Path(self._internal_path).rglob(pattern)]
655 else:
656 recursive_pattern = f"**/{pattern}"
657 return [
658 MultiStoragePath(str(p))
659 for p in self._storage_client.glob(
660 str(self._internal_path / recursive_pattern), include_url_prefix=True
661 )
662 ]
663
[docs]
664 def walk(self, top_down=True, on_error=None, follow_symlinks=False):
665 """
666 Walk the directory tree from this directory, similar to os.walk().
667
668 Not supported for remote storage paths.
669 """
670 if self._storage_client.is_default_profile():
671 return Path(self._internal_path).walk(top_down, on_error, follow_symlinks) # pyright: ignore[reportAttributeAccessIssue]
672 raise NotImplementedError("MultiStoragePath.walk() is unsupported for remote storage paths")
673
674 # Creating files and directories
675
[docs]
676 def touch(self, mode=0o666, exist_ok=False):
677 """
678 Create this file with the given access mode, if it doesn't exist.
679 """
680 if self._storage_client.is_default_profile():
681 Path(self._internal_path).touch(mode, exist_ok)
682 else:
683 if self.exists():
684 # object storage does not support updating the last modified time of an object without writing the object
685 logger.warning("MultiStoragePath.touch() is not supported for remote storage paths")
686 else:
687 self._storage_client.write(str(self._internal_path), b"")
688
[docs]
689 def mkdir(self, mode=0o777, parents=False, exist_ok=False) -> None:
690 """
691 Create a new directory at the given path.
692
693 For remote storage paths, this operation is a no-op.
694 """
695 if self._storage_client.is_default_profile():
696 Path(self._internal_path).mkdir(mode, parents, exist_ok)
697
[docs]
698 def symlink_to(self, target, target_is_directory=False):
699 """
700 Make this path a symlink pointing to the target path.
701
702 Not supported for remote storage paths.
703 """
704 if self._storage_client.is_default_profile():
705 Path(self._internal_path).symlink_to(target, target_is_directory)
706 else:
707 raise NotImplementedError("MultiStoragePath.symlink_to() is unsupported for remote storage paths")
708
709 # Renaming and deleting
710
[docs]
711 def rename(self, target) -> "MultiStoragePath":
712 """
713 Rename this path to the target path.
714 """
715 if not isinstance(target, MultiStoragePath):
716 target = MultiStoragePath(target)
717
718 if self._storage_client.is_default_profile():
719 Path(self._internal_path).rename(str(target._internal_path))
720 else:
721 # Note: This operation is not atomic, and the target path must be a single file.
722 self._storage_client.copy(str(self._internal_path), str(target._internal_path))
723 self._storage_client.delete(str(self._internal_path))
724
725 return target
726
[docs]
727 def replace(self, target):
728 """
729 Rename this path to the target path, overwriting if that path exists.
730
731 Not supported for remote storage paths.
732 """
733 if self._storage_client.is_default_profile():
734 Path(self._internal_path).replace(target)
735 else:
736 raise NotImplementedError("MultiStoragePath.replace() is unsupported for remote storage paths")
737
[docs]
738 def unlink(self, missing_ok: bool = False) -> None:
739 """
740 Remove this file or link. If the path is a directory, use rmdir() instead.
741 """
742 if self._storage_client.is_default_profile():
743 Path(self._internal_path).unlink(missing_ok=missing_ok)
744 else:
745 try:
746 self._storage_client.delete(str(self._internal_path))
747 except FileNotFoundError:
748 if not missing_ok:
749 raise
750
[docs]
751 def rmdir(self) -> None:
752 """
753 Remove this directory. The directory must be empty.
754
755 Not supported for remote storage paths.
756 """
757 if self._storage_client.is_default_profile():
758 Path(self._internal_path).rmdir()
759 else:
760 raise NotImplementedError("MultiStoragePath.rmdir() is unsupported for remote storage paths")
761
762 # Permissions and ownership
763
[docs]
764 def owner(self):
765 """
766 Return the login name of the file owner.
767
768 Not supported for remote storage paths.
769 """
770 if self._storage_client.is_default_profile():
771 return Path(self._internal_path).owner()
772 raise NotImplementedError("MultiStoragePath.owner() is unsupported for remote storage paths")
773
[docs]
774 def group(self):
775 """
776 Return the group name of the file gid.
777
778 Not supported for remote storage paths.
779 """
780 if self._storage_client.is_default_profile():
781 return Path(self._internal_path).group()
782 raise NotImplementedError("MultiStoragePath.group() is unsupported for remote storage paths")
783
[docs]
784 def chmod(self, mode):
785 """
786 Change the permissions of the path, like os.chmod().
787
788 Not supported for remote storage paths.
789 """
790 if self._storage_client.is_default_profile():
791 Path(self._internal_path).chmod(mode)
792 else:
793 raise NotImplementedError("MultiStoragePath.chmod() is unsupported for remote storage paths")
794
[docs]
795 def lchmod(self, mode):
796 """
797 Like chmod(), except if the path points to a symlink, the symlink's permissions are changed, rather
798 than its target's.
799
800 Not supported for remote storage paths.
801 """
802 if self._storage_client.is_default_profile():
803 Path(self._internal_path).lchmod(mode)
804 else:
805 raise NotImplementedError("MultiStoragePath.lchmod() is unsupported for remote storage paths")