_base.py 3.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134
  1. from __future__ import annotations
  2. import typing
  3. from abc import ABC, abstractmethod
  4. from ._copy import copy_dir, copy_file
  5. from ._errors import (
  6. DestinationExists,
  7. DirectoryExpected,
  8. FileExpected,
  9. FilesystemClosed,
  10. NoSysPath,
  11. ResourceNotFound,
  12. )
  13. from ._path import dirname
  14. from ._walk import BoundWalker
  15. if typing.TYPE_CHECKING:
  16. from typing import IO, Any, Collection, Iterator, Self, Type
  17. from ._info import Info
  18. from ._subfs import SubFS
  19. class FS(ABC):
  20. """Abstract base class for custom filesystems."""
  21. _closed: bool = False
  22. @abstractmethod
  23. def open(self, path: str, mode: str = "rb", **kwargs) -> IO[Any]: ...
  24. @abstractmethod
  25. def exists(self, path: str) -> bool: ...
  26. @abstractmethod
  27. def isdir(self, path: str) -> bool: ...
  28. @abstractmethod
  29. def isfile(self, path: str) -> bool: ...
  30. @abstractmethod
  31. def listdir(self, path: str) -> list[str]: ...
  32. @abstractmethod
  33. def makedir(self, path: str, recreate: bool = False) -> SubFS: ...
  34. @abstractmethod
  35. def makedirs(self, path: str, recreate: bool = False) -> SubFS: ...
  36. @abstractmethod
  37. def getinfo(self, path: str, namespaces: Collection[str] | None = None) -> Info: ...
  38. @abstractmethod
  39. def remove(self, path: str) -> None: ...
  40. @abstractmethod
  41. def removedir(self, path: str) -> None: ...
  42. @abstractmethod
  43. def removetree(self, path: str) -> None: ...
  44. @abstractmethod
  45. def movedir(self, src: str, dst: str, create: bool = False) -> None: ...
  46. def getsyspath(self, path: str) -> str:
  47. raise NoSysPath(f"the filesystem {self!r} has no system path")
  48. def close(self):
  49. self._closed = True
  50. def isclosed(self) -> bool:
  51. return self._closed
  52. def __enter__(self) -> Self:
  53. return self
  54. def __exit__(self, exc_type, exc, tb):
  55. self.close()
  56. return False # never swallow exceptions
  57. def check(self):
  58. if self._closed:
  59. raise FilesystemClosed(f"the filesystem {self!r} is closed")
  60. def opendir(self, path: str, *, factory: Type[SubFS] | None = None) -> SubFS:
  61. """Return a sub‑filesystem rooted at `path`."""
  62. if factory is None:
  63. from ._subfs import SubFS
  64. factory = SubFS
  65. return factory(self, path)
  66. def scandir(
  67. self, path: str, namespaces: Collection[str] | None = None
  68. ) -> Iterator[Info]:
  69. return (self.getinfo(f"{path}/{p}", namespaces) for p in self.listdir(path))
  70. @property
  71. def walk(self) -> BoundWalker:
  72. return BoundWalker(self)
  73. def readbytes(self, path: str) -> bytes:
  74. with self.open(path, "rb") as f:
  75. return f.read()
  76. def writebytes(self, path: str, data: bytes):
  77. with self.open(path, "wb") as f:
  78. f.write(data)
  79. def create(self, path: str, wipe: bool = False):
  80. if not wipe and self.exists(path):
  81. return False
  82. with self.open(path, "wb"):
  83. pass # 'touch' empty file
  84. return True
  85. def copy(self, src_path: str, dst_path: str, overwrite=False):
  86. if not self.exists(src_path):
  87. raise ResourceNotFound(f"{src_path!r} does not exist")
  88. elif not self.isfile(src_path):
  89. raise FileExpected(f"path {src_path!r} should be a file")
  90. if not overwrite and self.exists(dst_path):
  91. raise DestinationExists(f"destination {dst_path!r} already exists")
  92. if not self.isdir(dirname(dst_path)):
  93. raise DirectoryExpected(f"path {dirname(dst_path)!r} should be a directory")
  94. copy_file(self, src_path, self, dst_path)
  95. def copydir(self, src_path: str, dst_path: str, create=False):
  96. if not create and not self.exists(dst_path):
  97. raise ResourceNotFound(f"{dst_path!r} does not exist")
  98. if not self.isdir(src_path):
  99. raise DirectoryExpected(f"path {src_path!r} should be a directory")
  100. copy_dir(self, src_path, self, dst_path)