_osfs.py 5.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164
  1. from __future__ import annotations
  2. import errno
  3. import platform
  4. import shutil
  5. import stat
  6. import typing
  7. from os import PathLike
  8. from pathlib import Path
  9. from ._base import FS
  10. from ._errors import (
  11. CreateFailed,
  12. DirectoryExpected,
  13. DirectoryNotEmpty,
  14. FileExpected,
  15. IllegalDestination,
  16. ResourceError,
  17. ResourceNotFound,
  18. )
  19. from ._info import Info
  20. from ._path import isbase
  21. if typing.TYPE_CHECKING:
  22. from collections.abc import Collection
  23. from typing import IO, Any
  24. from ._subfs import SubFS
  25. _WINDOWS_PLATFORM = platform.system() == "Windows"
  26. class OSFS(FS):
  27. """Filesystem for a directory on the local disk.
  28. A thin layer on top of `pathlib.Path`.
  29. """
  30. def __init__(self, root: str | PathLike, create: bool = False):
  31. super().__init__()
  32. self._root = Path(root).resolve()
  33. if create:
  34. self._root.mkdir(parents=True, exist_ok=True)
  35. else:
  36. if not self._root.is_dir():
  37. raise CreateFailed(
  38. f"unable to create OSFS: {root!r} does not exist or is not a directory"
  39. )
  40. def _abs(self, rel_path: str) -> Path:
  41. self.check()
  42. return (self._root / rel_path.strip("/")).resolve()
  43. def open(self, path: str, mode: str = "rb", **kwargs) -> IO[Any]:
  44. try:
  45. return self._abs(path).open(mode, **kwargs)
  46. except FileNotFoundError:
  47. raise ResourceNotFound(f"No such file or directory: {path!r}")
  48. def exists(self, path: str) -> bool:
  49. return self._abs(path).exists()
  50. def isdir(self, path: str) -> bool:
  51. return self._abs(path).is_dir()
  52. def isfile(self, path: str) -> bool:
  53. return self._abs(path).is_file()
  54. def listdir(self, path: str) -> list[str]:
  55. return [p.name for p in self._abs(path).iterdir()]
  56. def _mkdir(self, path: str, parents: bool = False, exist_ok: bool = False) -> SubFS:
  57. self._abs(path).mkdir(parents=parents, exist_ok=exist_ok)
  58. return self.opendir(path)
  59. def makedir(self, path: str, recreate: bool = False) -> SubFS:
  60. return self._mkdir(path, parents=False, exist_ok=recreate)
  61. def makedirs(self, path: str, recreate: bool = False) -> SubFS:
  62. return self._mkdir(path, parents=True, exist_ok=recreate)
  63. def getinfo(self, path: str, namespaces: Collection[str] | None = None) -> Info:
  64. path = self._abs(path)
  65. if not path.exists():
  66. raise ResourceNotFound(f"No such file or directory: {str(path)!r}")
  67. info = {
  68. "basic": {
  69. "name": path.name,
  70. "is_dir": path.is_dir(),
  71. }
  72. }
  73. namespaces = namespaces or ()
  74. if "details" in namespaces:
  75. stat_result = path.stat()
  76. details = info["details"] = {
  77. "accessed": stat_result.st_atime,
  78. "modified": stat_result.st_mtime,
  79. "size": stat_result.st_size,
  80. "type": stat.S_IFMT(stat_result.st_mode),
  81. "created": getattr(stat_result, "st_birthtime", None),
  82. }
  83. ctime_key = "created" if _WINDOWS_PLATFORM else "metadata_changed"
  84. details[ctime_key] = stat_result.st_ctime
  85. return Info(info)
  86. def remove(self, path: str):
  87. path = self._abs(path)
  88. try:
  89. path.unlink()
  90. except FileNotFoundError:
  91. raise ResourceNotFound(f"No such file or directory: {str(path)!r}")
  92. except OSError as e:
  93. if path.is_dir():
  94. raise FileExpected(f"path {str(path)!r} should be a file")
  95. else:
  96. raise ResourceError(f"unable to remove {str(path)!r}: {e}")
  97. def removedir(self, path: str):
  98. try:
  99. self._abs(path).rmdir()
  100. except NotADirectoryError:
  101. raise DirectoryExpected(f"path {path!r} should be a directory")
  102. except OSError as e:
  103. if e.errno == errno.ENOTEMPTY:
  104. raise DirectoryNotEmpty(f"Directory not empty: {path!r}")
  105. else:
  106. raise ResourceError(f"unable to remove {path!r}: {e}")
  107. def removetree(self, path: str):
  108. shutil.rmtree(self._abs(path))
  109. def movedir(self, src_dir: str, dst_dir: str, create: bool = False):
  110. if isbase(src_dir, dst_dir):
  111. raise IllegalDestination(f"cannot move {src_dir!r} to {dst_dir!r}")
  112. src_path = self._abs(src_dir)
  113. if not src_path.exists():
  114. raise ResourceNotFound(f"Source {src_dir!r} does not exist")
  115. elif not src_path.is_dir():
  116. raise DirectoryExpected(f"Source {src_dir!r} should be a directory")
  117. dst_path = self._abs(dst_dir)
  118. if not create and not dst_path.exists():
  119. raise ResourceNotFound(f"Destination {dst_dir!r} does not exist")
  120. if dst_path.is_file():
  121. raise DirectoryExpected(f"Destination {dst_dir!r} should be a directory")
  122. if create:
  123. dst_path.parent.mkdir(parents=True, exist_ok=True)
  124. if dst_path.exists():
  125. if list(dst_path.iterdir()):
  126. raise DirectoryNotEmpty(f"Destination {dst_dir!r} is not empty")
  127. elif _WINDOWS_PLATFORM:
  128. # on Unix os.rename silently replaces an empty dst_dir whereas on
  129. # Windows it always raises FileExistsError, empty or not.
  130. dst_path.rmdir()
  131. src_path.rename(dst_path)
  132. def getsyspath(self, path: str) -> str:
  133. return str(self._abs(path))
  134. def __repr__(self) -> str:
  135. return f"{self.__class__.__name__}({str(self._root)!r})"
  136. def __str__(self) -> str:
  137. return f"<{self.__class__.__name__.lower()} '{self._root}'>"