lib: Add hierarchical FileType framework

Introduce FileType, a class consisting of a name and a parent property,
designed to make the information FileContext.stat() provides about a
path-addressable object richer.

A FileType property goes as ftype into the StatResult structure. Along with
it, this commit adds a central registry (FileTypes) where the type hierarchy
is kept, and additional parameters to put() and mkdir() to allow creating
typed FileContext objects.

Builtin types are:

  - FT_REG: A regular file, the default thing put() creates
  - FT_DIR: A directory, the default thing mkdir() creates
  - FT_TOP: The parent of the two
  - FT_COMPOUND: An FT_REG returning child entries from readdir()

Signed-off-by: Jan Lindemann <jan@janware.com>
Assisted-by: unsloth/Qwen3.8-27B-GGUF:Q4_K_M with pi.dev v1.0.1

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)
 

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(

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
+            )
+        )

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

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)