|
|
|
@ -79,6 +79,119 @@ DEFAULT_DIR_PERMS = Permissions(0o770) |
|
|
|
DEFAULT_FILE_PERMS = Permissions(0o640) |
|
|
|
|
|
|
|
|
|
|
|
FileType = Literal["dir", "file", "symlink", "other"] |
|
|
|
|
|
|
|
_IFMT_MAP: dict[int, FileType] = { |
|
|
|
S_IFDIR: "dir", |
|
|
|
S_IFREG: "file", |
|
|
|
S_IFLNK: "symlink", |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
@total_ordering |
|
|
|
class FileMetadata: |
|
|
|
"""Metadata associated with vfs files: file type, permissions, etc.""" |
|
|
|
|
|
|
|
path: PurePosixPath |
|
|
|
type: FileType |
|
|
|
permissions: Permissions |
|
|
|
modification_time: datetime |
|
|
|
byte_size: int |
|
|
|
_stat: stat_result |
|
|
|
|
|
|
|
def __init__( |
|
|
|
self, |
|
|
|
path: PurePosixPath, |
|
|
|
*, |
|
|
|
type: FileType, |
|
|
|
permissions: Permissions, |
|
|
|
modification_time: datetime, |
|
|
|
byte_size: int, |
|
|
|
): |
|
|
|
"""Create a `FileMetadata`.""" |
|
|
|
self.path = path |
|
|
|
self.type = type |
|
|
|
self.permissions = permissions |
|
|
|
self.modification_time = modification_time |
|
|
|
self.byte_size = byte_size |
|
|
|
|
|
|
|
@classmethod |
|
|
|
def from_stat( |
|
|
|
cls, |
|
|
|
path: PurePosixPath, |
|
|
|
stat: stat_result, |
|
|
|
) -> Self: |
|
|
|
"""Create a `FileMetadata` from a `stat_result`.""" |
|
|
|
return cls( |
|
|
|
path, |
|
|
|
type=_IFMT_MAP.get(S_IFMT(stat.st_mode), "other"), |
|
|
|
permissions=Permissions(S_IMODE(stat.st_mode)), |
|
|
|
modification_time=datetime.fromtimestamp(stat.st_mtime, UTC), |
|
|
|
byte_size=stat.st_size, |
|
|
|
) |
|
|
|
|
|
|
|
@property |
|
|
|
def unix_mode(self) -> str: |
|
|
|
"""Return unix-like mode in the form '-rwxrwxrwx'.""" |
|
|
|
return UNIX_MODE_FILE_TYPE[self.type] + str(self.permissions) |
|
|
|
|
|
|
|
@property |
|
|
|
def is_hidden(self) -> bool: |
|
|
|
"""Return true if the file starts with a '.'.""" |
|
|
|
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( |
|
|
|
self, |
|
|
|
) -> tuple[PurePosixPath, FileType, Permissions, datetime, int]: |
|
|
|
return ( |
|
|
|
self.path, |
|
|
|
self.type, |
|
|
|
self.permissions, |
|
|
|
self.modification_time, |
|
|
|
self.byte_size, |
|
|
|
) |
|
|
|
|
|
|
|
def __eq__(self, rhs: Any) -> bool: |
|
|
|
"""Test if two `Metadata` are the same.""" |
|
|
|
return ( |
|
|
|
self._as_tuple() == rhs._as_tuple() |
|
|
|
if isinstance(rhs, FileMetadata) |
|
|
|
else NotImplemented |
|
|
|
) |
|
|
|
|
|
|
|
def __lt__(self, rhs: Any) -> bool: |
|
|
|
"""Compare `rhs.path` with `self.path`.""" |
|
|
|
return self.path < rhs.path if isinstance(rhs, FileMetadata) else NotImplemented |
|
|
|
|
|
|
|
|
|
|
|
UNIX_MODE_FILE_TYPE = { |
|
|
|
"dir": "d", |
|
|
|
"file": "-", |
|
|
|
"other": "o", |
|
|
|
"link": "l", |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
class VirtualFileSystem: |
|
|
|
"""Represent a file system, with common file system operations.""" |
|
|
|
|
|
|
|
@ -127,7 +240,7 @@ class VirtualFileSystem: |
|
|
|
except OSError as err: |
|
|
|
msg = f"failed to read '{path}' metadata" |
|
|
|
raise FsError(msg) from err |
|
|
|
return FileMetadata.from_stat(self, path, stat) |
|
|
|
return FileMetadata.from_stat(path, stat) |
|
|
|
|
|
|
|
def iter_dir(self, path: AnyBsvPath) -> Iterator[FileMetadata]: |
|
|
|
"""Return the metadata of all items in the directory `path`.""" |
|
|
|
@ -136,7 +249,7 @@ class VirtualFileSystem: |
|
|
|
try: |
|
|
|
for entry in os.scandir(real_path): |
|
|
|
yield FileMetadata.from_stat( |
|
|
|
self, path / entry.name, entry.stat(follow_symlinks=False) |
|
|
|
path / entry.name, entry.stat(follow_symlinks=False) |
|
|
|
) |
|
|
|
except OSError as err: |
|
|
|
msg = f"failed to read directory {path}" |
|
|
|
@ -233,122 +346,3 @@ class VirtualFileSystem: |
|
|
|
|
|
|
|
def _real_path(self, path: PurePosixPath) -> Path: |
|
|
|
return self.path / path.relative_to("/") |
|
|
|
|
|
|
|
|
|
|
|
FileType = Literal["dir", "file", "symlink", "other"] |
|
|
|
|
|
|
|
_IFMT_MAP: dict[int, FileType] = { |
|
|
|
S_IFDIR: "dir", |
|
|
|
S_IFREG: "file", |
|
|
|
S_IFLNK: "symlink", |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
@total_ordering |
|
|
|
class FileMetadata: |
|
|
|
"""Metadata associated with vfs files: file type, permissions, etc.""" |
|
|
|
|
|
|
|
vfs: VirtualFileSystem |
|
|
|
path: PurePosixPath |
|
|
|
type: FileType |
|
|
|
permissions: Permissions |
|
|
|
modification_time: datetime |
|
|
|
byte_size: int |
|
|
|
_stat: stat_result |
|
|
|
|
|
|
|
def __init__( |
|
|
|
self, |
|
|
|
vfs: VirtualFileSystem, |
|
|
|
path: PurePosixPath, |
|
|
|
*, |
|
|
|
type: FileType, |
|
|
|
permissions: Permissions, |
|
|
|
modification_time: datetime, |
|
|
|
byte_size: int, |
|
|
|
): |
|
|
|
"""Create a `FileMetadata`.""" |
|
|
|
self.vfs = vfs |
|
|
|
self.path = path |
|
|
|
self.type = type |
|
|
|
self.permissions = permissions |
|
|
|
self.modification_time = modification_time |
|
|
|
self.byte_size = byte_size |
|
|
|
|
|
|
|
@classmethod |
|
|
|
def from_stat( |
|
|
|
cls, |
|
|
|
vfs: VirtualFileSystem, |
|
|
|
path: PurePosixPath, |
|
|
|
stat: stat_result, |
|
|
|
) -> Self: |
|
|
|
"""Create a `FileMetadata` from a `stat_result`.""" |
|
|
|
return cls( |
|
|
|
vfs, |
|
|
|
path, |
|
|
|
type=_IFMT_MAP.get(S_IFMT(stat.st_mode), "other"), |
|
|
|
permissions=Permissions(S_IMODE(stat.st_mode)), |
|
|
|
modification_time=datetime.fromtimestamp(stat.st_mtime, UTC), |
|
|
|
byte_size=stat.st_size, |
|
|
|
) |
|
|
|
|
|
|
|
@property |
|
|
|
def unix_mode(self) -> str: |
|
|
|
"""Return unix-like mode in the form '-rwxrwxrwx'.""" |
|
|
|
return UNIX_MODE_FILE_TYPE[self.type] + str(self.permissions) |
|
|
|
|
|
|
|
@property |
|
|
|
def is_hidden_files(self) -> bool: |
|
|
|
"""Return true if the file starts with a '.'.""" |
|
|
|
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( |
|
|
|
self, |
|
|
|
) -> tuple[VirtualFileSystem, PurePosixPath, FileType, Permissions, datetime, int]: |
|
|
|
return ( |
|
|
|
self.vfs, |
|
|
|
self.path, |
|
|
|
self.type, |
|
|
|
self.permissions, |
|
|
|
self.modification_time, |
|
|
|
self.byte_size, |
|
|
|
) |
|
|
|
|
|
|
|
def __eq__(self, rhs: Any) -> bool: |
|
|
|
"""Test if two `Metadata` are the same.""" |
|
|
|
return ( |
|
|
|
self._as_tuple() == rhs._as_tuple() |
|
|
|
if isinstance(rhs, FileMetadata) |
|
|
|
else NotImplemented |
|
|
|
) |
|
|
|
|
|
|
|
def __lt__(self, rhs: Any) -> bool: |
|
|
|
"""Compare `rhs.path` with `self.path`.""" |
|
|
|
return self.path < rhs.path if isinstance(rhs, FileMetadata) else NotImplemented |
|
|
|
|
|
|
|
|
|
|
|
UNIX_MODE_FILE_TYPE = { |
|
|
|
"dir": "d", |
|
|
|
"file": "-", |
|
|
|
"other": "o", |
|
|
|
"link": "l", |
|
|
|
} |
|
|
|
|