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