| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164 |
- from __future__ import annotations
- import errno
- import platform
- import shutil
- import stat
- import typing
- from os import PathLike
- from pathlib import Path
- from ._base import FS
- from ._errors import (
- CreateFailed,
- DirectoryExpected,
- DirectoryNotEmpty,
- FileExpected,
- IllegalDestination,
- ResourceError,
- ResourceNotFound,
- )
- from ._info import Info
- from ._path import isbase
- if typing.TYPE_CHECKING:
- from collections.abc import Collection
- from typing import IO, Any
- from ._subfs import SubFS
- _WINDOWS_PLATFORM = platform.system() == "Windows"
- class OSFS(FS):
- """Filesystem for a directory on the local disk.
- A thin layer on top of `pathlib.Path`.
- """
- def __init__(self, root: str | PathLike, create: bool = False):
- super().__init__()
- self._root = Path(root).resolve()
- if create:
- self._root.mkdir(parents=True, exist_ok=True)
- else:
- if not self._root.is_dir():
- raise CreateFailed(
- f"unable to create OSFS: {root!r} does not exist or is not a directory"
- )
- def _abs(self, rel_path: str) -> Path:
- self.check()
- return (self._root / rel_path.strip("/")).resolve()
- def open(self, path: str, mode: str = "rb", **kwargs) -> IO[Any]:
- try:
- return self._abs(path).open(mode, **kwargs)
- except FileNotFoundError:
- raise ResourceNotFound(f"No such file or directory: {path!r}")
- def exists(self, path: str) -> bool:
- return self._abs(path).exists()
- def isdir(self, path: str) -> bool:
- return self._abs(path).is_dir()
- def isfile(self, path: str) -> bool:
- return self._abs(path).is_file()
- def listdir(self, path: str) -> list[str]:
- return [p.name for p in self._abs(path).iterdir()]
- def _mkdir(self, path: str, parents: bool = False, exist_ok: bool = False) -> SubFS:
- self._abs(path).mkdir(parents=parents, exist_ok=exist_ok)
- return self.opendir(path)
- def makedir(self, path: str, recreate: bool = False) -> SubFS:
- return self._mkdir(path, parents=False, exist_ok=recreate)
- def makedirs(self, path: str, recreate: bool = False) -> SubFS:
- return self._mkdir(path, parents=True, exist_ok=recreate)
- def getinfo(self, path: str, namespaces: Collection[str] | None = None) -> Info:
- path = self._abs(path)
- if not path.exists():
- raise ResourceNotFound(f"No such file or directory: {str(path)!r}")
- info = {
- "basic": {
- "name": path.name,
- "is_dir": path.is_dir(),
- }
- }
- namespaces = namespaces or ()
- if "details" in namespaces:
- stat_result = path.stat()
- details = info["details"] = {
- "accessed": stat_result.st_atime,
- "modified": stat_result.st_mtime,
- "size": stat_result.st_size,
- "type": stat.S_IFMT(stat_result.st_mode),
- "created": getattr(stat_result, "st_birthtime", None),
- }
- ctime_key = "created" if _WINDOWS_PLATFORM else "metadata_changed"
- details[ctime_key] = stat_result.st_ctime
- return Info(info)
- def remove(self, path: str):
- path = self._abs(path)
- try:
- path.unlink()
- except FileNotFoundError:
- raise ResourceNotFound(f"No such file or directory: {str(path)!r}")
- except OSError as e:
- if path.is_dir():
- raise FileExpected(f"path {str(path)!r} should be a file")
- else:
- raise ResourceError(f"unable to remove {str(path)!r}: {e}")
- def removedir(self, path: str):
- try:
- self._abs(path).rmdir()
- except NotADirectoryError:
- raise DirectoryExpected(f"path {path!r} should be a directory")
- except OSError as e:
- if e.errno == errno.ENOTEMPTY:
- raise DirectoryNotEmpty(f"Directory not empty: {path!r}")
- else:
- raise ResourceError(f"unable to remove {path!r}: {e}")
- def removetree(self, path: str):
- shutil.rmtree(self._abs(path))
- def movedir(self, src_dir: str, dst_dir: str, create: bool = False):
- if isbase(src_dir, dst_dir):
- raise IllegalDestination(f"cannot move {src_dir!r} to {dst_dir!r}")
- src_path = self._abs(src_dir)
- if not src_path.exists():
- raise ResourceNotFound(f"Source {src_dir!r} does not exist")
- elif not src_path.is_dir():
- raise DirectoryExpected(f"Source {src_dir!r} should be a directory")
- dst_path = self._abs(dst_dir)
- if not create and not dst_path.exists():
- raise ResourceNotFound(f"Destination {dst_dir!r} does not exist")
- if dst_path.is_file():
- raise DirectoryExpected(f"Destination {dst_dir!r} should be a directory")
- if create:
- dst_path.parent.mkdir(parents=True, exist_ok=True)
- if dst_path.exists():
- if list(dst_path.iterdir()):
- raise DirectoryNotEmpty(f"Destination {dst_dir!r} is not empty")
- elif _WINDOWS_PLATFORM:
- # on Unix os.rename silently replaces an empty dst_dir whereas on
- # Windows it always raises FileExistsError, empty or not.
- dst_path.rmdir()
- src_path.rename(dst_path)
- def getsyspath(self, path: str) -> str:
- return str(self._abs(path))
- def __repr__(self) -> str:
- return f"{self.__class__.__name__}({str(self._root)!r})"
- def __str__(self) -> str:
- return f"<{self.__class__.__name__.lower()} '{self._root}'>"
|