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