165 lines
5.6 KiB
Python
165 lines
5.6 KiB
Python
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}'>"
|