Merge pull request #1 from effigies/rf/upath

rf: Replace direntry with UPath in FileTree

Author: ubdbra001

src/bids_validator/context.py

@@ -61,7 +61,7 @@ def datatype_to_modality(datatype: str, schema: Namespace) -> str:
 @cache
 def load_tsv(file: FileTree, *, max_rows=0) -> Namespace:
     """Load TSV contents into a Namespace."""
-    with open(file) as fobj:
+    with file.path_obj.open() as fobj:
         if max_rows > 0:
             fobj = itertools.islice(fobj, max_rows)
         contents = (line.rstrip('\r\n').split('\t') for line in fobj)
@@ -72,7 +72,7 @@ def load_tsv(file: FileTree, *, max_rows=0) -> Namespace:
 @cache
 def load_json(file: FileTree) -> dict[str]:
     """Load JSON file contents."""
-    return orjson.loads(UPath(file).read_bytes())
+    return orjson.loads(file.path_obj.read_bytes())
 
 
 class Subjects:
@@ -360,7 +360,7 @@ def modality(self) -> str | None:
     @property
     def size(self) -> int:
         """Length of the current file in bytes."""
-        return self.file.direntry.stat().st_size
+        return self.file.path_obj.stat().st_size
 
     @property
     def associations(self) -> ctx.Associations:

src/bids_validator/types/files.py

@@ -4,103 +4,50 @@
 
 import os
 import posixpath
-import stat
 from functools import cached_property
 from pathlib import Path
 
 import attrs
+from upath import UPath
 
 from . import _typings as t
 
 __all__ = ('FileTree',)
 
 
-@attrs.define
-class UserDirEntry:
-    """Partial reimplementation of :class:`os.DirEntry`.
-
-    :class:`os.DirEntry` can't be instantiated from Python, but this can.
-    """
-
-    path: str = attrs.field(repr=False, converter=os.fspath)
-    name: str = attrs.field(init=False)
-    _stat: os.stat_result = attrs.field(init=False, repr=False, default=None)
-    _lstat: os.stat_result = attrs.field(init=False, repr=False, default=None)
-
-    def __attrs_post_init__(self) -> None:
-        self.name = os.path.basename(self.path)
-
-    def __fspath__(self) -> str:
-        return self.path
-
-    def stat(self, *, follow_symlinks: bool = True) -> os.stat_result:
-        """Return stat_result object for the entry; cached per entry."""
-        if follow_symlinks:
-            if self._stat is None:
-                self._stat = os.stat(self.path, follow_symlinks=True)
-            return self._stat
-        else:
-            if self._lstat is None:
-                self._lstat = os.stat(self.path, follow_symlinks=False)
-            return self._lstat
-
-    def is_dir(self, *, follow_symlinks: bool = True) -> bool:
-        """Return True if the entry is a directory; cached per entry."""
-        _stat = self.stat(follow_symlinks=follow_symlinks)
-        return stat.S_ISDIR(_stat.st_mode)
-
-    def is_file(self, *, follow_symlinks: bool = True) -> bool:
-        """Return True if the entry is a file; cached per entry."""
-        _stat = self.stat(follow_symlinks=follow_symlinks)
-        return stat.S_ISREG(_stat.st_mode)
-
-    def is_symlink(self) -> bool:
-        """Return True if the entry is a symlink; cached per entry."""
-        _stat = self.stat(follow_symlinks=False)
-        return stat.S_ISLNK(_stat.st_mode)
-
-
-def as_direntry(obj: os.PathLike) -> os.DirEntry | UserDirEntry:
-    """Convert PathLike into DirEntry-like object."""
-    if isinstance(obj, os.DirEntry):
-        return obj
-    return UserDirEntry(obj)
-
-
-@attrs.define
+@attrs.define(frozen=True)
 class FileTree:
     """Represent a FileTree with cached metadata."""
 
-    direntry: os.DirEntry | UserDirEntry = attrs.field(repr=False, converter=as_direntry)
-    parent: FileTree | None = attrs.field(repr=False, default=None)
-    is_dir: bool = attrs.field(default=False)
-    children: dict[str, FileTree] = attrs.field(repr=False, factory=dict)
-    name: str = attrs.field(init=False)
+    path_obj: UPath = attrs.field(repr=False, converter=UPath)
+    is_dir: bool = attrs.field(repr=False, default=None)
+    parent: FileTree | None = attrs.field(repr=False, default=None, eq=False)
+    children: dict[str, FileTree] = attrs.field(repr=False, factory=dict, eq=False)
 
     def __attrs_post_init__(self):
-        self.name = self.direntry.name
-        self.children = {
-            name: attrs.evolve(child, parent=self) for name, child in self.children.items()
-        }
+        if self.is_dir is None:
+            object.__setattr__(self, 'is_dir', self.path_obj.is_dir())
+        object.__setattr__(
+            self,
+            'children',
+            {name: attrs.evolve(child, parent=self) for name, child in self.children.items()},
+        )
 
     @classmethod
-    def read_from_filesystem(
-        cls,
-        direntry: os.PathLike,
-        parent: FileTree | None = None,
-    ) -> t.Self:
-        """Read a FileTree from the filesystem.
-
-        Uses :func:`os.scandir` to walk the directory tree.
-        """
-        self = cls(direntry, parent=parent)
-        if self.direntry.is_dir():
-            self.is_dir = True
-            self.children = {
-                entry.name: FileTree.read_from_filesystem(entry, parent=self)
-                for entry in os.scandir(self.direntry)
+    def read_from_filesystem(cls, path_obj: os.PathLike) -> t.Self:
+        """Read a FileTree from the filesystem."""
+        path_obj = UPath(path_obj)
+        children = {}
+        if is_dir := path_obj.is_dir():
+            children = {
+                entry.name: FileTree.read_from_filesystem(entry) for entry in path_obj.iterdir()
             }
-        return self
+        return cls(path_obj, is_dir=is_dir, children=children)
+
+    @property
+    def name(self) -> bool:
+        """The name of the current FileTree node."""
+        return self.path_obj.name
 
     def __contains__(self, relpath: os.PathLike) -> bool:
         parts = Path(relpath).parts
@@ -110,10 +57,7 @@ def __contains__(self, relpath: os.PathLike) -> bool:
         return child and (len(parts) == 1 or posixpath.join(*parts[1:]) in child)
 
     def __fspath__(self):
-        return self.direntry.path
-
-    def __hash__(self):
-        return hash(self.direntry.path)
+        return self.path_obj.__fspath__()
 
     def __truediv__(self, relpath: str | os.PathLike) -> t.Self:
         parts = Path(relpath).parts

tests/test_context.py

@@ -1,3 +1,6 @@
+import json
+
+import fsspec
 import pytest
 from bidsschematools.types.context import Subject
 
@@ -10,6 +13,14 @@ def synthetic_dataset(examples):
     return FileTree.read_from_filesystem(examples / 'synthetic')
 
 
+@pytest.fixture
+def memfs():
+    mem = fsspec.filesystem('memory')
+    mem.store.clear()
+    yield mem
+    mem.store.clear()
+
+
 def test_load(synthetic_dataset, schema):
     ds = context.Dataset(synthetic_dataset, schema)
 
@@ -115,6 +126,33 @@ def test_sidecar_inheritance(examples):
     assert sidecar['RepetitionTimePreparation'] == 5.5
 
 
+def test_sidecar_order(memfs):
+    """Test to ensure inheritance principle is skipped when inherit=False"""
+    root_json = {'rootOverwriteA': 'root', 'rootOverwriteB': 'root', 'rootValue': 'root'}
+    subject_json = {'rootOverwriteA': 'subject', 'subOverwrite': 'subject', 'subValue': 'subject'}
+    anat_json = {'rootOverwriteB': 'anat', 'subOverwrite': 'anat', 'anatValue': 'anat'}
+    memfs.pipe(
+        {
+            '/T1w.json': json.dumps(root_json).encode(),
+            '/sub-01/sub-01_T1w.json': json.dumps(subject_json).encode(),
+            '/sub-01/anat/sub-01_T1w.json': json.dumps(anat_json).encode(),
+            '/sub-01/anat/sub-01_T1w.nii': b'',
+        }
+    )
+
+    dataset = FileTree.read_from_filesystem('memory://')
+    file = dataset / 'sub-01' / 'anat' / 'sub-01_T1w.nii'
+    sidecar = context.load_sidecar(file)
+    assert sidecar == {
+        'rootValue': 'root',
+        'subValue': 'subject',
+        'rootOverwriteA': 'subject',
+        'anatValue': 'anat',
+        'rootOverwriteB': 'anat',
+        'subOverwrite': 'anat',
+    }
+
+
 def test_sessions(synthetic_dataset):
     sub01 = synthetic_dataset / 'sub-01'