|
|
@ -95,20 +95,39 @@ class VirtualFileSystem: |
|
|
|
|
|
|
|
|
def is_file(self, path: AnyBsvPath) -> bool: |
|
|
def is_file(self, path: AnyBsvPath) -> bool: |
|
|
"""Test if `path` is a file.""" |
|
|
"""Test if `path` is a file.""" |
|
|
path = self._make_path(path) |
|
|
return self.metadata(path).is_file |
|
|
return self._real_path(path).is_file() |
|
|
|
|
|
|
|
|
|
|
|
def is_dir(self, path: AnyBsvPath) -> bool: |
|
|
def is_dir(self, path: AnyBsvPath) -> bool: |
|
|
"""Test if `path` is a directory.""" |
|
|
"""Test if `path` is a directory.""" |
|
|
path = self._make_path(path) |
|
|
return self.metadata(path).is_dir |
|
|
return self._real_path(path).is_dir() |
|
|
|
|
|
|
|
|
def is_symlink(self, path: AnyBsvPath) -> bool: |
|
|
|
|
|
"""Test if `path` is a symbolic link.""" |
|
|
|
|
|
return self.metadata(path).is_symlink |
|
|
|
|
|
|
|
|
|
|
|
def is_other(self, path: AnyBsvPath) -> bool: |
|
|
|
|
|
"""Test if `path` is not a file, directory or symbolic link.""" |
|
|
|
|
|
return self.metadata(path).is_other |
|
|
|
|
|
|
|
|
def metadata(self, path: AnyBsvPath) -> FileMetadata: |
|
|
def metadata(self, path: AnyBsvPath) -> FileMetadata: |
|
|
"""Return the metadata of a given object.""" |
|
|
"""Return the metadata of a given object.""" |
|
|
|
|
|
metadata = self.metadata_or_none(path) |
|
|
|
|
|
if metadata is None: |
|
|
|
|
|
msg = f"file '{path}' not found" |
|
|
|
|
|
raise NotFoundError(msg) |
|
|
|
|
|
return metadata |
|
|
|
|
|
|
|
|
|
|
|
def metadata_or_none(self, path: AnyBsvPath) -> FileMetadata | None: |
|
|
|
|
|
"""Return the metadata of a given object or `None` if it does not exists.""" |
|
|
path = self._make_path(path) |
|
|
path = self._make_path(path) |
|
|
return FileMetadata.from_stat( |
|
|
try: |
|
|
self, path, self._real_path(path).stat(follow_symlinks=False) |
|
|
stat = self._real_path(path).stat(follow_symlinks=False) |
|
|
) |
|
|
except FileNotFoundError: |
|
|
|
|
|
return None |
|
|
|
|
|
except OSError as err: |
|
|
|
|
|
msg = f"failed to read '{path}' metadata" |
|
|
|
|
|
raise FsError(msg) from err |
|
|
|
|
|
return FileMetadata.from_stat(self, path, stat) |
|
|
|
|
|
|
|
|
def iter_dir(self, path: AnyBsvPath) -> Iterator[FileMetadata]: |
|
|
def iter_dir(self, path: AnyBsvPath) -> Iterator[FileMetadata]: |
|
|
"""Return the metadata of all items in the directory `path`.""" |
|
|
"""Return the metadata of all items in the directory `path`.""" |
|
|
@ -125,31 +144,18 @@ class VirtualFileSystem: |
|
|
|
|
|
|
|
|
def read_bytes(self, path: AnyBsvPath) -> bytes: |
|
|
def read_bytes(self, path: AnyBsvPath) -> bytes: |
|
|
"""Return the content of `path` as `bytes`.""" |
|
|
"""Return the content of `path` as `bytes`.""" |
|
|
path = self._make_path(path) |
|
|
with self.open_read(path) as stream: |
|
|
try: |
|
|
return stream.read() |
|
|
return self._real_path(path).read_bytes() |
|
|
|
|
|
except OSError as err: |
|
|
|
|
|
msg = f"failed to read {path}" |
|
|
|
|
|
raise FsError(msg) from err |
|
|
|
|
|
|
|
|
|
|
|
def write_bytes(self, path: AnyBsvPath, data: Buffer | BinaryIO) -> int: |
|
|
def write_bytes(self, path: AnyBsvPath, data: Buffer | BinaryIO) -> int: |
|
|
"""Create or replace a file at `path`, setting its content to `data`.""" |
|
|
"""Create or replace a file at `path`, setting its content to `data`.""" |
|
|
path = self._make_path(path) |
|
|
|
|
|
real_path = self._real_path(path) |
|
|
|
|
|
written = 0 |
|
|
written = 0 |
|
|
|
|
|
with self.open_write(path) as sout: |
|
|
try: |
|
|
|
|
|
stream = real_path.open("wb") |
|
|
|
|
|
except OSError as err: |
|
|
|
|
|
msg = f"failed to write {path}" |
|
|
|
|
|
raise FsError(msg) from err |
|
|
|
|
|
|
|
|
|
|
|
with stream: |
|
|
|
|
|
if isinstance(data, Buffer): |
|
|
if isinstance(data, Buffer): |
|
|
written += stream.write(data) |
|
|
written += sout.write(data) |
|
|
else: |
|
|
else: |
|
|
while chunk := data.read(65536): |
|
|
while chunk := data.read(65536): |
|
|
written += stream.write(chunk) |
|
|
written += sout.write(chunk) |
|
|
return written |
|
|
return written |
|
|
|
|
|
|
|
|
def open_read(self, path: AnyBsvPath) -> BinaryIO: |
|
|
def open_read(self, path: AnyBsvPath) -> BinaryIO: |
|
|
@ -229,12 +235,12 @@ class VirtualFileSystem: |
|
|
return self.path / path.relative_to("/") |
|
|
return self.path / path.relative_to("/") |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
FileType = Literal["dir", "file", "link", "other"] |
|
|
FileType = Literal["dir", "file", "symlink", "other"] |
|
|
|
|
|
|
|
|
_IFMT_MAP = { |
|
|
_IFMT_MAP: dict[int, FileType] = { |
|
|
S_IFDIR: "dir", |
|
|
S_IFDIR: "dir", |
|
|
S_IFREG: "file", |
|
|
S_IFREG: "file", |
|
|
S_IFLNK: "link", |
|
|
S_IFLNK: "symlink", |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@ -279,7 +285,7 @@ class FileMetadata: |
|
|
return cls( |
|
|
return cls( |
|
|
vfs, |
|
|
vfs, |
|
|
path, |
|
|
path, |
|
|
type=cast("FileType", _IFMT_MAP.get(S_IFMT(stat.st_mode), "other")), |
|
|
type=_IFMT_MAP.get(S_IFMT(stat.st_mode), "other"), |
|
|
permissions=Permissions(S_IMODE(stat.st_mode)), |
|
|
permissions=Permissions(S_IMODE(stat.st_mode)), |
|
|
modification_time=datetime.fromtimestamp(stat.st_mtime, UTC), |
|
|
modification_time=datetime.fromtimestamp(stat.st_mtime, UTC), |
|
|
byte_size=stat.st_size, |
|
|
byte_size=stat.st_size, |
|
|
@ -295,6 +301,26 @@ class FileMetadata: |
|
|
"""Return true if the file starts with a '.'.""" |
|
|
"""Return true if the file starts with a '.'.""" |
|
|
return self.path.name.startswith(".") |
|
|
return self.path.name.startswith(".") |
|
|
|
|
|
|
|
|
|
|
|
@property |
|
|
|
|
|
def is_file(self) -> bool: |
|
|
|
|
|
"""Test if this is a file.""" |
|
|
|
|
|
return self.type == "file" |
|
|
|
|
|
|
|
|
|
|
|
@property |
|
|
|
|
|
def is_dir(self) -> bool: |
|
|
|
|
|
"""Test if this is a directory.""" |
|
|
|
|
|
return self.type == "dir" |
|
|
|
|
|
|
|
|
|
|
|
@property |
|
|
|
|
|
def is_symlink(self) -> bool: |
|
|
|
|
|
"""Test if this is a symbolic link.""" |
|
|
|
|
|
return self.type == "symlink" |
|
|
|
|
|
|
|
|
|
|
|
@property |
|
|
|
|
|
def is_other(self) -> bool: |
|
|
|
|
|
"""Test if this is a symbolic link.""" |
|
|
|
|
|
return self.type == "other" |
|
|
|
|
|
|
|
|
def _as_tuple( |
|
|
def _as_tuple( |
|
|
self, |
|
|
self, |
|
|
) -> tuple[VirtualFileSystem, PurePosixPath, FileType, Permissions, datetime, int]: |
|
|
) -> tuple[VirtualFileSystem, PurePosixPath, FileType, Permissions, datetime, int]: |
|
|
|