_subfs.py 3.0 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192
  1. from __future__ import annotations
  2. import typing
  3. from pathlib import PurePosixPath
  4. from ._base import FS
  5. from ._errors import DirectoryExpected, ResourceNotFound
  6. if typing.TYPE_CHECKING:
  7. from collections.abc import Collection
  8. from typing import IO, Any
  9. from ._info import Info
  10. class SubFS(FS):
  11. """Maps a sub-directory of another filesystem."""
  12. def __init__(self, parent: FS, sub_path: str):
  13. super().__init__()
  14. self._parent = parent
  15. self._prefix = PurePosixPath(sub_path).as_posix().rstrip("/")
  16. if not parent.exists(self._prefix):
  17. raise ResourceNotFound(f"No such file or directory: {sub_path!r}")
  18. elif not parent.isdir(self._prefix):
  19. raise DirectoryExpected(f"{sub_path!r} is not a directory")
  20. def delegate_fs(self):
  21. return self._parent
  22. def _full(self, rel: str) -> str:
  23. self.check()
  24. return f"{self._prefix}/{PurePosixPath(rel).as_posix()}".lstrip("/")
  25. def open(self, path: str, mode: str = "rb", **kwargs) -> IO[Any]:
  26. return self._parent.open(self._full(path), mode, **kwargs)
  27. def exists(self, path: str) -> bool:
  28. return self._parent.exists(self._full(path))
  29. def isdir(self, path: str) -> bool:
  30. return self._parent.isdir(self._full(path))
  31. def isfile(self, path: str) -> bool:
  32. return self._parent.isfile(self._full(path))
  33. def listdir(self, path: str) -> list[str]:
  34. return self._parent.listdir(self._full(path))
  35. def makedir(self, path: str, recreate: bool = False):
  36. return self._parent.makedir(self._full(path), recreate=recreate)
  37. def makedirs(self, path: str, recreate: bool = False):
  38. return self._parent.makedirs(self._full(path), recreate=recreate)
  39. def getinfo(self, path: str, namespaces: Collection[str] | None = None) -> Info:
  40. return self._parent.getinfo(self._full(path), namespaces=namespaces)
  41. def remove(self, path: str):
  42. return self._parent.remove(self._full(path))
  43. def removedir(self, path: str):
  44. return self._parent.removedir(self._full(path))
  45. def removetree(self, path: str):
  46. return self._parent.removetree(self._full(path))
  47. def movedir(self, src: str, dst: str, create: bool = False):
  48. self._parent.movedir(self._full(src), self._full(dst), create=create)
  49. def getsyspath(self, path: str) -> str:
  50. return self._parent.getsyspath(self._full(path))
  51. def readbytes(self, path: str) -> bytes:
  52. return self._parent.readbytes(self._full(path))
  53. def writebytes(self, path: str, data: bytes):
  54. self._parent.writebytes(self._full(path), data)
  55. def __repr__(self) -> str:
  56. return f"{self.__class__.__name__}({self._parent!r}, {self._prefix!r})"
  57. def __str__(self) -> str:
  58. return f"{self._parent}/{self._prefix}"
  59. class ClosingSubFS(SubFS):
  60. """Like SubFS, but auto-closes the parent filesystem when closed."""
  61. def close(self):
  62. super().close()
  63. self._parent.close()