diff --git a/src/python/jw/pkg/lib/ExecContext.py b/src/python/jw/pkg/lib/ExecContext.py index 9c137cc9..a282dc62 100644 --- a/src/python/jw/pkg/lib/ExecContext.py +++ b/src/python/jw/pkg/lib/ExecContext.py @@ -10,6 +10,7 @@ from typing import TYPE_CHECKING, Any, Iterable, NamedTuple, override if TYPE_CHECKING: from typing import Type from types import TracebackType + from .file_types import FileType from .base import Input, InputMode, Result, StatResult from .FileContext import FileContext as Base @@ -500,6 +501,7 @@ class ExecContext(Base): group: str | None, mode: str | None, atomic: bool, + ftype: FileType, ) -> Result: from .util import pretty_cmd @@ -584,7 +586,7 @@ class ExecContext(Base): await self.run(cmd, cmd_input = InputMode.NonInteractive) @override - async def _mkdir(self, path: str, mode: int) -> None: + async def _mkdir(self, path: str, mode: int, ftype: FileType) -> None: cmd = ['mkdir', path, '-m', self.__mode_str(mode)] await self.run(cmd, cmd_input = InputMode.NonInteractive) diff --git a/src/python/jw/pkg/lib/FileContext.py b/src/python/jw/pkg/lib/FileContext.py index c16a5b69..e452d871 100644 --- a/src/python/jw/pkg/lib/FileContext.py +++ b/src/python/jw/pkg/lib/FileContext.py @@ -6,12 +6,14 @@ from enum import Enum, auto from functools import cached_property from typing import TYPE_CHECKING, Any, Iterable, override +from .file_types import FT_DIR, FT_REG from .log import DEBUG, ERR, log from .ProcFilter import ProcPipeline from .Uri import Uri if TYPE_CHECKING: from .base import Result, StatResult + from .file_types import FileType from .ProcFilter import ProcFilter class FileContext(abc.ABC): @@ -161,6 +163,7 @@ class FileContext(abc.ABC): group: str | None, mode: str | None, atomic: bool, + ftype: FileType, ) -> Result: raise NotImplementedError() @@ -176,6 +179,7 @@ class FileContext(abc.ABC): group: str | None = None, mode: int | None = None, atomic: bool = False, + ftype: FileType = FT_REG, ) -> Result: mode_str = None if mode is None else oct(mode).replace('0o', '0') if self.__out_pipe is not None: @@ -191,6 +195,7 @@ class FileContext(abc.ABC): group = group, mode = mode_str, atomic = atomic, + ftype = ftype, ) async def _unlink(self, path: str) -> None: @@ -217,11 +222,13 @@ class FileContext(abc.ABC): async def rename(self, src: str, dst: str) -> None: return await self._rename(src, dst) - async def _mkdir(self, path: str, mode: int) -> None: + async def _mkdir(self, path: str, mode: int, ftype: FileType) -> None: raise NotImplementedError(f'{self.log_name}: mkdir({path}) is not implemented') - async def mkdir(self, path: str, mode: int = 0o777) -> None: - return await self._mkdir(self._chroot(path), mode) + async def mkdir( + self, path: str, mode: int = 0o777, ftype: FileType = FT_DIR + ) -> None: + return await self._mkdir(self._chroot(path), mode, ftype) async def _mktemp(self, tmpl: str, directory: bool) -> str: raise NotImplementedError( diff --git a/src/python/jw/pkg/lib/base.py b/src/python/jw/pkg/lib/base.py index 2e050731..d297158e 100644 --- a/src/python/jw/pkg/lib/base.py +++ b/src/python/jw/pkg/lib/base.py @@ -1,8 +1,9 @@ from __future__ import annotations from enum import Enum, auto -from typing import TYPE_CHECKING, NamedTuple, TypeAlias +from typing import TYPE_CHECKING, Self, TypeAlias, override +from .file_types import FT_COMPOUND, FT_DIR, FT_REG, FileType, file_types from .Result import Result as Result if TYPE_CHECKING: @@ -16,20 +17,47 @@ class InputMode(Enum): Input: TypeAlias = InputMode | bytes | str -class StatResult(NamedTuple): - mode: int - owner: str - group: str - size: int - atime: float - mtime: float - ctime: float +class StatResult: + + __slots__ = ('mode', 'owner', 'group', 'size', 'atime', 'mtime', 'ctime', 'ftype') + + def __init__( + self, + mode: int, + owner: str, + group: str, + size: int, + atime: float, + mtime: float, + ctime: float, + ftype: str | FileType | None = None, + ) -> None: + self.mode = mode + self.owner = owner + self.group = group + self.size = size + self.atime = atime + self.mtime = mtime + self.ctime = ctime + self.ftype = file_types.get(mode if ftype is None else ftype) + + @property + def is_dir(self) -> bool: + # -- A directory is what you can descend into: a branch, + # and a compound, which is a file that lists its keys + return self.ftype.is_subtype(FT_DIR) or self.ftype.is_subtype(FT_COMPOUND) + + @property + def is_gettable(self) -> bool: + return self.ftype.is_subtype(FT_REG) @classmethod - def from_os(cls, rhs: os.stat_result) -> StatResult: + def from_os(cls, rhs: os.stat_result) -> Self: + import grp import pwd - return StatResult( + + return cls( rhs.st_mode, pwd.getpwuid(rhs.st_uid).pw_name, grp.getgrgid(rhs.st_gid).gr_name, @@ -38,3 +66,40 @@ class StatResult(NamedTuple): rhs.st_mtime, rhs.st_ctime, ) + + @override + def __repr__(self) -> str: + return ( + f'StatResult(mode={self.mode}, owner={self.owner!r}, ' + f'group={self.group!r}, size={self.size}, ' + f'atime={self.atime}, mtime={self.mtime}, ctime={self.ctime}, ' + f'ftype={self.ftype!r})' + ) + + @override + def __eq__(self, rhs: object) -> bool: + + if not isinstance(rhs, StatResult): + return NotImplemented + + return ( + self.mode == rhs.mode and self.owner == rhs.owner + and self.group == rhs.group and self.size == rhs.size + and self.atime == rhs.atime and self.mtime == rhs.mtime + and self.ctime == rhs.ctime and self.ftype == rhs.ftype + ) + + @override + def __hash__(self) -> int: + return hash( + ( + self.mode, + self.owner, + self.group, + self.size, + self.atime, + self.mtime, + self.ctime, + self.ftype + ) + ) diff --git a/src/python/jw/pkg/lib/ec/Local.py b/src/python/jw/pkg/lib/ec/Local.py index b4c530ff..24e55642 100644 --- a/src/python/jw/pkg/lib/ec/Local.py +++ b/src/python/jw/pkg/lib/ec/Local.py @@ -16,6 +16,7 @@ from ..log import ERR, NOTICE, log if TYPE_CHECKING: from typing import Any, Callable + from ..file_types import FileType from ..Uri import Uri class Local(Base): @@ -180,7 +181,7 @@ class Local(Base): os.rename(src, dst) @override - async def _mkdir(self, path: str, mode: int) -> None: + async def _mkdir(self, path: str, mode: int, ftype: FileType) -> None: os.mkdir(path, mode) @override diff --git a/src/python/jw/pkg/lib/file_types.py b/src/python/jw/pkg/lib/file_types.py new file mode 100644 index 00000000..845d1a78 --- /dev/null +++ b/src/python/jw/pkg/lib/file_types.py @@ -0,0 +1,127 @@ +from __future__ import annotations + +import stat + +from typing import TYPE_CHECKING, override + +if TYPE_CHECKING: + from typing import Mapping + +FT_NAME_TOP = 'top' +FT_NAME_REG = 'reg' +FT_NAME_DIR = 'dir' + +class FileType: + + __slots__ = ('name', 'parent') + + name: str + parent: FileType + + def __init__(self, name: str, parent: FileType | None) -> None: + self.name = name + self.parent = parent if parent is not None else self + + @override + def __repr__(self) -> str: + return self.name + + def is_subtype(self, rhs: FileType) -> bool: + cur = self + while True: + if cur.name == rhs.name: + return True + if cur == cur.parent: + return False + cur = cur.parent + +class FileTypes: + + class Error(ValueError): + pass + + # -- Public API + + def __init__(self) -> None: + self.__types: dict[str, FileType] = {FT_NAME_TOP: FileType(FT_NAME_TOP, None)} + self.register_types({ + FT_NAME_DIR: FT_NAME_TOP, + FT_NAME_REG: FT_NAME_TOP, + }) + + def register( + self, + new_type: str | FileType, + parent: str | FileType | None = None, + ) -> FileType: + if parent is None: + if isinstance(new_type, str): + raise self.Error( + f'Tried to register file type {new_type} without parent' + ) + elif isinstance(new_type, FileType): + parent = new_type.parent + parent_name = parent if isinstance(parent, str) else parent.name + parent_type = self.__types.get(parent_name) + if parent_type is None: + raise self.Error( + f'Tried to register file type {new_type} with ' + f'inexistent parent {parent}' + ) + if isinstance(parent, FileType) and parent != parent_type: + raise self.Error( + f'Tried to register file type {new_type} with ' + f'parent {parent} differing from registered parent' + ) + old = self.__types.get(new_type if isinstance(new_type, str) else new_type.name) + if old is not None: + if old.parent != parent_type: + raise self.Error( + f'Tried to register existing file type {new_type} -> ' + f'{old.parent} with new parent {parent_type}' + ) + return old + ret = ( + new_type + if isinstance(new_type, FileType) else FileType(new_type, parent_type) + ) + self.__types[ret.name] = ret + return ret + + def register_types(self, types: Mapping[str, str | FileType]) -> None: + + def __pop_type(names: set[str]) -> FileType: + for name in names: + parent = types[name] + p = self.__types.get(parent if isinstance(parent, str) else parent.name) + if p is not None: + names.remove(name) + return FileType(name, p) + raise self.Error(f'Unregistered parent in file types: {names}') + + new: set[str] = set(types.keys()) + while new: + self.register(__pop_type(new)) + + @classmethod + def from_mode(cls, mode: int) -> FileType: + if stat.S_ISDIR(mode): + return FT_DIR + if stat.S_ISREG(mode): + return FT_REG + raise NotImplementedError(f'Unsupported file type in mode {oct(mode)}') + + def get(self, ftype: str | FileType | int) -> FileType: + if isinstance(ftype, FileType): + return ftype + if isinstance(ftype, int): + return self.from_mode(ftype) + return self.__types[ftype] + +file_types = FileTypes() + +FT_TOP = file_types.get(FT_NAME_TOP) +FT_REG = file_types.get(FT_NAME_REG) +FT_DIR = file_types.get(FT_NAME_DIR) + +FT_COMPOUND = file_types.register('compound', FT_REG)