From 88e7867db84328408576da1c5cb924e7429fab81 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 6 Oct 2026 20:51:13 +0000 Subject: [PATCH 01/10] validate: stream missing annexed content with datalad-fuse Add a `stream` policy to `--missing-file-content`: the content of annexed files that is not present locally (as in a DataLad clone of a Dandiset) is streamed with datalad-fuse's adapter, so that pynwb and nwbinspector run on it without it being downloaded. pynwb validation results are cached under the files' git-annex keys. This is the validation integration of #1933, on top of the datalad-fuse readable rather than a git-only reader of the git-annex branch. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Qn5WBSiQgoZoL4fytF6nEr --- dandi/cli/cmd_validate.py | 9 +- dandi/cli/tests/test_cmd_validate.py | 43 +++++++- dandi/files/bases.py | 82 ++++++++++++-- dandi/files/bids.py | 2 +- dandi/pynwb_utils.py | 51 +++++++-- dandi/validate/_core.py | 135 +++++++++++++++++------ dandi/validate/_types.py | 6 + dandi/validate/tests/test_core.py | 158 ++++++++++++++++++++++++++- docs/source/cmdline/validate.rst | 62 ++++++++++- 9 files changed, 495 insertions(+), 53 deletions(-) diff --git a/dandi/cli/cmd_validate.py b/dandi/cli/cmd_validate.py index 5241c5eb4..ea18707c0 100644 --- a/dandi/cli/cmd_validate.py +++ b/dandi/cli/cmd_validate.py @@ -170,8 +170,13 @@ def _filter_results( "in a datalad dataset without fetched data). 'error' (default) emits a " "concise error per file, 'skip' skips each such file with a warning, " "'only-non-data' skips content-dependent validators but still validates " - "path layout.", - type=click.Choice(["error", "only-non-data", "skip"], case_sensitive=True), + "path layout, 'stream' streams the content of annexed files with " + "datalad-fuse so that content-dependent validators run without the files " + "having to be downloaded (requires git-annex and `pip install " + "'dandi[datalad]'`).", + type=click.Choice( + ["error", "only-non-data", "skip", "stream"], case_sensitive=True + ), default="error", ) @click.option( diff --git a/dandi/cli/tests/test_cmd_validate.py b/dandi/cli/tests/test_cmd_validate.py index 1da5c16ca..aef651740 100644 --- a/dandi/cli/tests/test_cmd_validate.py +++ b/dandi/cli/tests/test_cmd_validate.py @@ -1,6 +1,7 @@ import json from pathlib import Path -from typing import cast +import shutil +from typing import Any, cast from click.testing import CliRunner import pytest @@ -998,6 +999,46 @@ def test_validate_missing_file_content_no_broken_symlinks(tmp_path: Path) -> Non assert "FILE_CONTENT_MISSING" not in r_skip.output +@pytest.mark.ai_generated +def test_validate_missing_file_content_stream( + tmp_path: Path, simple3_nwb: Path, range_http_server: tuple[Any, str] +) -> None: + """--missing-file-content=stream validates annexed content with datalad-fuse.""" + from ...tests.fixtures import make_git_annex_dandiset + from ...tests.skip import skipif + + skipif.no_git_annex() + pytest.importorskip("datalad_fuse.fsspec") + _, base_url = range_http_server + shutil.copy(simple3_nwb, tmp_path / "served" / "content.nwb") + ds = tmp_path / "ds" + make_git_annex_dandiset( + ds, {"sub-001/sub-001.nwb": (simple3_nwb, f"{base_url}/content.nwb")} + ) + out = tmp_path / "out.jsonl" + r = CliRunner().invoke( + validate, + [ + "--missing-file-content", + "stream", + "--min-severity", + "INFO", + "-f", + "json_lines", + "-o", + str(out), + str(ds), + ], + ) + assert "Traceback" not in r.output + ids = [rec.id for rec in load_validation_jsonl([str(out)])] + assert "DANDI.FILE_CONTENT_STREAMED" in ids + assert "DANDI.FILE_CONTENT_MISSING" not in ids + # simple3_nwb lacks a subject_id, which only content-based checks notice + assert "NWBI.check_subject_id_exists" in ids + assert r.exit_code == 1 + + @pytest.mark.ai_generated def test_validate_broken_symlink_path_argument(tmp_path: Path) -> None: """A broken symlink can be given directly as a path to validate. diff --git a/dandi/files/bases.py b/dandi/files/bases.py index f2d0f22cd..4bbc6dc5f 100644 --- a/dandi/files/bases.py +++ b/dandi/files/bases.py @@ -10,6 +10,7 @@ from pathlib import Path import re from threading import Lock +import traceback from typing import IO, Any, Generic from xml.etree.ElementTree import fromstring @@ -32,7 +33,7 @@ set_asset_schema_key, ) from dandi.metadata.core import get_default_metadata -from dandi.misctypes import DUMMY_DANDI_ETAG, Digest, LocalReadableFile, P +from dandi.misctypes import DUMMY_DANDI_ETAG, Digest, LocalReadableFile, P, Readable from dandi.utils import post_upload_size_check, pre_upload_size_check, yaml_load from dandi.validate._types import ( ORIGIN_INTERNAL_DANDI, @@ -320,12 +321,22 @@ class LocalFileAsset(LocalAsset): an asset of a Dandiset """ + #: An alternative source for the asset's content, used instead of the file + #: at `filepath` when set. This is used to stream the content of an + #: annexed file (e.g., in a DataLad dataset) that is not present locally; + #: see `dandi.support.datalad_fuse`. + content_source: Readable | None = None + + def _content(self) -> Path | Readable: + """The source to read the asset's content from""" + return self.content_source if self.content_source is not None else self.filepath + def get_metadata( self, digest: Digest | None = None, ignore_errors: bool = True, ) -> BareAsset: - metadata = get_default_metadata(self.filepath, digest=digest) + metadata = get_default_metadata(self._content(), digest=digest) metadata.path = self.path return metadata @@ -515,7 +526,7 @@ def get_metadata( from dandi.metadata.nwb import nwb2asset try: - metadata = nwb2asset(self.filepath, digest=digest) + metadata = nwb2asset(self._content(), digest=digest) except Exception as e: lgr.warning( "Failed to extract NWB metadata from %s: %s: %s", @@ -524,7 +535,7 @@ def get_metadata( str(e), ) if ignore_errors: - metadata = get_default_metadata(self.filepath, digest=digest) + metadata = get_default_metadata(self._content(), digest=digest) else: raise metadata.path = self.path @@ -555,12 +566,16 @@ def get_validation_errors( pass else: # Avoid heavy import by importing within function: - from nwbinspector import Importance, inspect_nwbfile, load_config + from nwbinspector import Importance, load_config # Avoid heavy import by importing within function: from dandi.pynwb_utils import validate as pynwb_validate - errors.extend(pynwb_validate(self.filepath, devel_debug=devel_debug)) + errors.extend( + pynwb_validate( + self.filepath, devel_debug=devel_debug, readable=self.content_source + ) + ) if schema_version is not None: errors.extend( super().get_validation_errors( @@ -576,9 +591,7 @@ def get_validation_errors( validator_version=str(_get_nwb_inspector_version()), ) - for error in inspect_nwbfile( - nwbfile_path=self.filepath, - skip_validate=True, + for error in self._inspect_nwbfile( config=load_config(filepath_or_keyword="dandi"), importance_threshold=Importance.BEST_PRACTICE_VIOLATION, ): @@ -623,6 +636,57 @@ def get_validation_errors( ) return errors + def _inspect_nwbfile(self, **kwargs: Any) -> Iterator[Any]: + """ + Yield the messages from inspecting the file with nwbinspector, reading + the content from `content_source` if it is set (in which case what + `nwbinspector.inspect_nwbfile()` does for a local file is replicated + here for the streamed content) + """ + # Avoid heavy import by importing within function: + from nwbinspector import ( + Importance, + InspectorMessage, + inspect_nwbfile, + inspect_nwbfile_object, + ) + + if self.content_source is None: + yield from inspect_nwbfile( + nwbfile_path=self.filepath, skip_validate=True, **kwargs + ) + return + + # Avoid heavy import by importing within function: + import h5py + from pynwb import NWBHDF5IO + + from dandi.pynwb_utils import open_readable + + try: + with open_readable(self.content_source) as fp, h5py.File( + fp, "r" + ) as h5, NWBHDF5IO(file=h5, mode="r", load_namespaces=True) as io: + nwbfile = io.read() + for message in inspect_nwbfile_object(nwbfile_object=nwbfile, **kwargs): + if message is not None: + message.file_path = str(self.filepath) + yield message + except Exception as e: + # Report the failure the same way inspect_nwbfile() does for a + # local file that PyNWB cannot read + yield InspectorMessage( + message=traceback.format_exc(), + importance=Importance.ERROR, + check_function_name=( + f"During io.read(), an error occurred: " + f"{type(e).__module__}.{type(e).__name__}. " + "This indicates that PyNWB was unable to read the file. " + "See the traceback message for more details." + ), + file_path=str(self.filepath), + ) + class VideoAsset(LocalFileAsset): pass diff --git a/dandi/files/bids.py b/dandi/files/bids.py index 371f7cd78..27823b296 100644 --- a/dandi/files/bids.py +++ b/dandi/files/bids.py @@ -262,7 +262,7 @@ def get_bids_metadata(self, digest: Digest | None = None) -> BareAsset: """ metadata = self.bids_dataset_description.get_asset_metadata(self) start_time = end_time = datetime.now().astimezone() - add_common_metadata(metadata, self.filepath, start_time, end_time, digest) + add_common_metadata(metadata, self._content(), start_time, end_time, digest) metadata.path = self.path return metadata diff --git a/dandi/pynwb_utils.py b/dandi/pynwb_utils.py index b0dda71c0..c065dad77 100644 --- a/dandi/pynwb_utils.py +++ b/dandi/pynwb_utils.py @@ -611,8 +611,9 @@ def rename_if_matched(_name: str, obj: Any) -> None: f.visititems(rename_if_matched) -@validate_cache.memoize_path -def validate(path: str | Path, devel_debug: bool = False) -> list[ValidationResult]: +def validate( + path: str | Path, devel_debug: bool = False, *, readable: Readable | None = None +) -> list[ValidationResult]: """Run validation on a file and return errors In case of an exception being thrown, an error message added to the @@ -621,8 +622,38 @@ def validate(path: str | Path, devel_debug: bool = False) -> list[ValidationResu Parameters ---------- path: str or Path + Path of the file, as reported in the returned results + devel_debug: bool, optional + Whether to re-raise exceptions instead of reporting them as errors + readable: Readable, optional + If given, the file's content is read from this `Readable` instead of from + ``path`` (e.g., to stream the content of an annexed file which is not + present locally); the results are then cached only if it is an + `AnnexedReadableFile` (see `annex_fingerprint`). + """ + source: str | Readable = readable if readable is not None else str(path) + # fscacher is untyped, so the memoized _validate returns Any for mypy + return cast( + "list[ValidationResult]", + _validate(source, str(path), devel_debug=devel_debug), + ) + + +@validate_cache.memoize_path(custom_fingerprint=annex_fingerprint) +def _validate( + source: str | Readable, path: str, devel_debug: bool = False +) -> list[ValidationResult]: + """`validate` proper, with the content source as first argument for caching + + Parameters + ---------- + source: str or Readable + What to read the file's content from: its path or a `Readable` + path: str + Path of the file, as reported in the returned results + devel_debug: bool + Whether to re-raise exceptions instead of reporting them as errors """ - path = str(path) # Might come in as pathlib's PATH errors: list[ValidationResult] = [] # To overcome @@ -634,7 +665,7 @@ def validate(path: str | Path, devel_debug: bool = False) -> list[ValidationResu ) version = None try: - version = get_nwb_version(path, sanitize=False) + version = get_nwb_version(source, sanitize=False) except Exception: # we just will not remove any errors, it is required so should be some pass @@ -680,9 +711,15 @@ def validate(path: str | Path, devel_debug: bool = False) -> list[ValidationResu ) try: - # Validates against the namespaces cached in the file; pynwb falls back - # to its own namespaces if the file has none cached - error_outputs = pynwb.validate(path=path) + # Either way, validates against the namespaces cached in the file; pynwb + # falls back to its own namespaces if the file has none cached + if isinstance(source, Readable): + with open_readable(source) as fp, h5py.File(fp, "r") as h5, NWBHDF5IO( + file=h5, mode="r", load_namespaces=True + ) as reader: + error_outputs = pynwb.validate(io=reader) + else: + error_outputs = pynwb.validate(path=source) except Exception as exc: if devel_debug: raise diff --git a/dandi/validate/_core.py b/dandi/validate/_core.py index 371ce5a30..e3c0e4f33 100644 --- a/dandi/validate/_core.py +++ b/dandi/validate/_core.py @@ -10,9 +10,10 @@ from __future__ import annotations from collections.abc import Iterator +from importlib.util import find_spec import os from pathlib import Path -from typing import Any +import shutil from ._types import ( ORIGIN_VALIDATION_DANDI_LAYOUT, @@ -26,7 +27,8 @@ Validator, ) from ..consts import dandiset_metadata_file -from ..files import find_dandi_files +from ..files import DandiFile, LocalFileAsset, find_dandi_files +from ..support.datalad_fuse import AnnexedReadableFile, get_annexed_readable from ..utils import find_parent_directory_containing BIDS_TO_DANDI = { @@ -155,7 +157,8 @@ def _is_broken_symlink(filepath: Path) -> bool: # BIDS error codes that require reading file content (headers, pixel data). -# When ``only-non-data`` is active these are suppressed for broken-symlink files. +# When ``only-non-data`` or ``stream`` is active these are suppressed for +# broken-symlink files (the BIDS validator cannot stream content). _BIDS_CONTENT_DEPENDENT_CODES = frozenset( { "BIDS.NIFTI_HEADER_UNREADABLE", @@ -181,13 +184,20 @@ def validate( Policy for files whose content is unavailable (e.g. broken symlinks in a datalad dataset without fetched data). ``error`` emits a concise error, ``skip`` skips the file with a warning, ``only-non-data`` skips - content-dependent validators but still validates path layout. + content-dependent validators but still validates path layout, and + ``stream`` streams the content of annexed files with datalad-fuse (see + `dandi.support.datalad_fuse`) so that + content-dependent validators run without the content being present + locally. Yields ------ path, errors errors for a path """ + if missing_file_content == MissingFileContent.stream: + _check_streaming_requirements() + # Archive of unique `ValidationResult` objects obtained through # `DandiFile.get_validation_errors()` # Note: This is needed to hold on to the unique `ValidationResult` objects @@ -214,34 +224,43 @@ def validate( p, dandiset_path=dandiset_path, allow_all=allow_any_path ): # Handle broken symlinks (missing file content) - if _is_broken_symlink(df.filepath): - r = _handle_missing_content(df, missing_file_content) - if r is not None: - r_id = id(r) - if r_id not in df_result_ids: - df_results.append(r) - df_result_ids.add(r_id) - yield r - if missing_file_content in ( - MissingFileContent.skip, - MissingFileContent.error, - ): + is_broken = _is_broken_symlink(df.filepath) + if is_broken: + if missing_file_content == MissingFileContent.stream: + readable = _prepare_streaming(df) + # A file whose content cannot be streamed gets an error + # and is otherwise skipped + skip_file = readable is None + else: + readable = None + skip_file = missing_file_content in ( + MissingFileContent.skip, + MissingFileContent.error, + ) + r = _handle_missing_content(df, missing_file_content, readable) + r_id = id(r) + if r_id not in df_result_ids: + df_results.append(r) + df_result_ids.add(r_id) + yield r + if skip_file: continue - # only-non-data: fall through but pass the flag to validators + # only-non-data & stream: fall through but pass the policy to + # the validators - is_broken = _is_broken_symlink(df.filepath) for r in df.get_validation_errors( schema_version=schema_version, devel_debug=devel_debug, missing_file_content=(missing_file_content if is_broken else None), ): - # For broken-symlink files under only-non-data, suppress - # BIDS errors that require reading file content (e.g. + # For broken-symlink files under only-non-data and stream, + # suppress BIDS errors that require reading file content (e.g. # NIFTI_HEADER_UNREADABLE). The validator ran in full so # real files still get those checks. if ( is_broken - and missing_file_content == MissingFileContent.only_non_data + and missing_file_content + in (MissingFileContent.only_non_data, MissingFileContent.stream) and r.id in _BIDS_CONTENT_DEPENDENT_CODES ): continue @@ -252,21 +271,74 @@ def validate( yield r +def _check_streaming_requirements() -> None: + """Raise an informative error if what is needed for streaming file content + is not installed""" + if find_spec("datalad_fuse") is None: + raise RuntimeError( + "Streaming file content requires datalad-fuse; install it with " + "`pip install 'dandi[datalad]'`" + ) + if shutil.which("git-annex") is None: + raise RuntimeError("Streaming file content requires git-annex") + + +def _prepare_streaming(df: DandiFile) -> AnnexedReadableFile | None: + """ + Set up streaming of the content of the annexed file represented by ``df``, + and return the `Readable` that will be used for it, or `None` if the + content cannot be streamed (see `get_annexed_readable()`) + """ + if not isinstance(df, LocalFileAsset): + return None + readable = get_annexed_readable(df.filepath) + if readable is not None: + df.content_source = readable + return readable + + def _handle_missing_content( - df: Any, + df: DandiFile, policy: MissingFileContent, -) -> ValidationResult | None: + readable: AnnexedReadableFile | None = None, +) -> ValidationResult: """Produce a single :class:`ValidationResult` for a file with missing content. - Returns ``None`` when *policy* is ``only-non-data`` (a warning is not - needed because validation still proceeds on the non-data aspects). + For the ``stream`` policy, ``readable`` is the `Readable` streaming the + file's content, or `None` if the content cannot be streamed. """ - from ..files import DandiFile - - assert isinstance(df, DandiFile) filepath = df.filepath - if policy == MissingFileContent.error: + if policy == MissingFileContent.stream: + if readable is not None: + return ValidationResult( + id="DANDI.FILE_CONTENT_STREAMED", + origin=ORIGIN_VALIDATION_DANDI_LAYOUT, + severity=Severity.INFO, + scope=Scope.FILE, + path=filepath, + dandiset_path=df.dandiset_path, + message=( + f"File content is not present locally (git-annex key " + f"{readable.key}); content-dependent validation streams " + f"it from {readable.url}" + ), + ) + return ValidationResult( + id="DANDI.FILE_CONTENT_MISSING", + origin=ORIGIN_VALIDATION_DANDI_LAYOUT, + severity=Severity.ERROR, + scope=Scope.FILE, + path=filepath, + dandiset_path=df.dandiset_path, + message=( + f"File content is not available (broken symlink: " + f"{filepath} -> {os.readlink(filepath)}) and cannot be " + f"streamed: not an annexed file in a git-annex repository " + f"with a URL known to git-annex." + ), + ) + elif policy == MissingFileContent.error: return ValidationResult( id="DANDI.FILE_CONTENT_MISSING", origin=ORIGIN_VALIDATION_DANDI_LAYOUT, @@ -277,8 +349,9 @@ def _handle_missing_content( message=( f"File content is not available (broken symlink: " f"{filepath} -> {os.readlink(filepath)}). " - f"Use --missing-file-content=skip or " - f"--missing-file-content=only-non-data to handle gracefully." + f"Use --missing-file-content=skip, " + f"--missing-file-content=only-non-data, or " + f"--missing-file-content=stream to handle gracefully." ), ) elif policy == MissingFileContent.skip: diff --git a/dandi/validate/_types.py b/dandi/validate/_types.py index f41c8402e..48b18b45a 100644 --- a/dandi/validate/_types.py +++ b/dandi/validate/_types.py @@ -26,6 +26,12 @@ class MissingFileContent(StrEnum): skip = auto() """Skip the file entirely; emit a WARNING noting that validation was skipped.""" + stream = auto() + """Stream the content of annexed files with datalad-fuse (see + `dandi.support.datalad_fuse`) so that content-dependent validators run + without the content being present locally. An INFO result is emitted + for each streamed file, and an ERROR for each file that cannot be streamed.""" + lgr = logging.getLogger(__name__) diff --git a/dandi/validate/tests/test_core.py b/dandi/validate/tests/test_core.py index 7d331bb61..ab3759542 100644 --- a/dandi/validate/tests/test_core.py +++ b/dandi/validate/tests/test_core.py @@ -1,5 +1,7 @@ import json +import os from pathlib import Path +import shutil from typing import Any import pytest @@ -17,7 +19,14 @@ ) from ... import __version__ from ...consts import dandiset_metadata_file -from ...tests.fixtures import BIDS_TESTDATA_SELECTION +from ...pynwb_utils import validate as pynwb_validate +from ...support.datalad_fuse import get_annexed_readable +from ...tests.fixtures import ( + BIDS_TESTDATA_SELECTION, + RangeHTTPServer, + make_git_annex_dandiset, +) +from ...tests.skip import skipif def test_validate_nwb_error(simple3_nwb: Path) -> None: @@ -310,3 +319,150 @@ def test_validate_broken_symlink_real_file_still_validated(tmp_path: Path) -> No assert ( len(broken_pynwb) == 0 ), f"policy={policy.value}: pynwb should not run on the broken symlink" + + +# ---- Tests for streaming the content of annexed files (missing_file_content=stream) ---- + + +@pytest.fixture +def streamable_dandiset( + range_http_server: tuple[RangeHTTPServer, str], simple3_nwb: Path, tmp_path: Path +) -> tuple[Path, RangeHTTPServer]: + """A git-annex dandiset looking like a DataLad clone without fetched content. + + ``sub-001/sub-001.nwb`` is an annexed copy of *simple3_nwb* whose content + is served by the HTTP server, and ``sub-001/sub-001_video.mp4`` an annexed + video registered at a URL that is never read (only the size recorded in + its key is needed). + """ + skipif.no_git_annex() + pytest.importorskip("datalad_fuse.fsspec") + server, base_url = range_http_server + shutil.copy(simple3_nwb, tmp_path / "served" / "content.nwb") + video = tmp_path / "video.mp4" + video.write_bytes(b"not really a video") + ds = tmp_path / "ds" + make_git_annex_dandiset( + ds, + { + "sub-001/sub-001.nwb": (simple3_nwb, f"{base_url}/content.nwb"), + "sub-001/sub-001_video.mp4": (video, f"{base_url}/never-read.mp4"), + }, + ) + return ds, server + + +def _content_results(results: list[ValidationResult], name: str) -> list[tuple]: + """Path-independent summary of the results for the file called *name*.""" + return sorted( + (r.id, r.severity, r.origin.validator, r.message) + for r in results + if r.path is not None + and r.path.name == name + and r.id != "DANDI.FILE_CONTENT_STREAMED" + ) + + +@pytest.mark.ai_generated +def test_validate_stream( + streamable_dandiset: tuple[Path, RangeHTTPServer], + simple3_nwb: Path, + tmp_path: Path, +) -> None: + """stream policy validates the content of annexed files with datalad-fuse.""" + ds, server = streamable_dandiset + results = list(validate(ds, missing_file_content=MissingFileContent.stream)) + + streamed = [r for r in results if r.id == "DANDI.FILE_CONTENT_STREAMED"] + assert sorted(r.path.name for r in streamed if r.path is not None) == [ + "sub-001.nwb", + "sub-001_video.mp4", + ] + assert all(r.severity == Severity.INFO for r in streamed) + assert not [r for r in results if r.id.startswith("DANDI.FILE_CONTENT_MISSING")] + assert server.ranged_requests > 0 + + # Validating the video only needs the size recorded in its key + assert _content_results(results, "sub-001_video.mp4") == [] + + # Content-dependent validation of the NWB file gives the same results as + # for a regular dandiset containing the file itself + local = tmp_path / "local" + (local / "sub-001").mkdir(parents=True) + shutil.copy(ds / dandiset_metadata_file, local / dandiset_metadata_file) + shutil.copy(simple3_nwb, local / "sub-001" / "sub-001.nwb") + expected = list(validate(local)) + # simple3_nwb lacks a subject_id, which only content-based checks notice + assert any(r.origin.validator == Validator.nwbinspector for r in expected) + assert _content_results(results, "sub-001.nwb") == _content_results( + expected, "sub-001.nwb" + ) + + +@pytest.mark.ai_generated +@pytest.mark.skipif( + os.environ.get("DANDI_CACHE") == "ignore", reason="the validation cache is disabled" +) +def test_validate_stream_cached_by_key( + streamable_dandiset: tuple[Path, RangeHTTPServer], +) -> None: + """pynwb validation results of a streamed file are cached under its annex key.""" + ds, server = streamable_dandiset + nwb = ds / "sub-001" / "sub-001.nwb" + readable = get_annexed_readable(nwb) + assert readable is not None + results = pynwb_validate(nwb, readable=readable) + assert not [r for r in results if r.id == "pynwb.GENERIC"] + requests = server.ranged_requests + assert requests > 0 + # Served from the cache, without the content being read again + assert pynwb_validate(nwb, readable=get_annexed_readable(nwb)) == results + assert server.ranged_requests == requests + + +@pytest.mark.ai_generated +def test_validate_stream_not_streamable(tmp_path: Path) -> None: + """stream policy emits an error for a broken symlink that cannot be streamed.""" + pytest.importorskip("datalad_fuse.fsspec") + skipif.no_git_annex() + ds = _make_dandiset_with_broken_symlink(tmp_path) + results = list(validate(ds, missing_file_content=MissingFileContent.stream)) + errs = [r for r in results if r.id == "DANDI.FILE_CONTENT_MISSING"] + assert len(errs) == 1 + assert errs[0].severity == Severity.ERROR + assert errs[0].message is not None + assert "cannot be streamed" in errs[0].message + assert not [r for r in results if r.id == "DANDI.FILE_CONTENT_STREAMED"] + assert not [ + r + for r in results + if r.origin.validator in (Validator.pynwb, Validator.nwbinspector) + ] + + +@pytest.mark.ai_generated +def test_validate_stream_unreadable_url( + streamable_dandiset: tuple[Path, RangeHTTPServer], tmp_path: Path +) -> None: + """A registered URL that cannot be read yields errors, not an exception.""" + ds, _ = streamable_dandiset + (tmp_path / "served" / "content.nwb").unlink() + results = list(validate(ds, missing_file_content=MissingFileContent.stream)) + nwb = ds / "sub-001" / "sub-001.nwb" + assert [ + r for r in results if r.path == nwb and r.id == "DANDI.FILE_CONTENT_STREAMED" + ] + errs = [r for r in results if r.path == nwb and r.severity == Severity.ERROR] + assert {r.origin.validator for r in errs} == { + Validator.pynwb, + Validator.nwbinspector, + } + + +@pytest.mark.ai_generated +def test_validate_stream_requires_datalad_fuse( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr("dandi.validate._core.find_spec", lambda name: None) + with pytest.raises(RuntimeError, match=r"dandi\[datalad\]"): + list(validate(tmp_path, missing_file_content=MissingFileContent.stream)) diff --git a/docs/source/cmdline/validate.rst b/docs/source/cmdline/validate.rst index 3a07d7e7a..a789f6203 100644 --- a/docs/source/cmdline/validate.rst +++ b/docs/source/cmdline/validate.rst @@ -56,7 +56,7 @@ Options Limit the number of results shown per group (or in total when not grouping); the excess is replaced by a count of omitted results -.. option:: --missing-file-content [error|only-non-data|skip] +.. option:: --missing-file-content [error|only-non-data|skip|stream] How to handle files whose content is unavailable, such as the broken symbolic links of a DataLad_ dataset (a git-annex_ repository) whose @@ -73,6 +73,12 @@ Options Skip content-dependent validators (pynwb, nwbinspector, ...) for each such file but still validate its path layout + ``stream`` + Stream the content of each such file with datalad-fuse_ so that + content-dependent validators run without the file having to be + downloaded; see `Validating DataLad Dandisets Remotely`_ + below. + .. option:: --load Instead of running validation, load previously saved results from the @@ -84,6 +90,60 @@ Options .. _git-annex: https://git-annex.branchable.com +Validating DataLad Dandisets Remotely +------------------------------------- + +Every Dandiset on the DANDI Archive is mirrored as a DataLad dataset at +https://github.com/dandisets (with https://github.com/dandisets/dandisets as +the superdataset containing all of them). In such a dataset, the assets are +annexed files: symbolic links that remain broken until the content is fetched, +which for some Dandisets would mean downloading terabytes of data. The +``stream`` policy of :option:`--missing-file-content` lets ``dandi validate`` +run the content-dependent validators (pynwb and nwbinspector for NWB files) on +such files by streaming their content on demand with datalad-fuse_ (without +mounting anything), which asks git-annex where the content is, reading only +the parts of each file that the validators need. This requires git-annex and +``pip install "dandi[datalad]"``. + +For example, to produce a JSON Lines record of *all* validation results for +a Dandiset:: + + datalad clone https://github.com/dandisets/000003 + dandi validate --missing-file-content=stream --min-severity=INFO \ + --format=json_lines --output=000003.jsonl 000003 + +Each streamed file yields an ``INFO``-level ``DANDI.FILE_CONTENT_STREAMED`` +result naming the URL its content was read from; a file that cannot be streamed +(not an annexed file, or no URL known to git-annex) yields a +``DANDI.FILE_CONTENT_MISSING`` error instead. + +Notes: + +- git-annex must be initialized in the clone, which ``datalad clone`` does + (after ``git clone``, run ``git annex init``). +- Some nwbinspector checks read data arrays (e.g., timestamps), so the amount + of data streamed for a file depends on its content; it is nevertheless + usually a small fraction of the file. +- The pynwb validation results and the metadata of a streamed file are cached + under its git-annex key (a digest of its content), just as those of a local + file are cached under its modification time and size, so re-running the + command skips that work for files whose key has not changed since. As for + local files, set the :envvar:`DANDI_CACHE` environment variable to ``clear`` + to start afresh or to ``ignore`` to bypass the cache. +- Zarr assets are stored as separate subdatasets (https://github.com/dandizarrs) + and are not streamed yet: an uninstalled Zarr subdataset is an empty directory + that is not validated at all. Streaming them is a follow-up for when NWB Zarr + support has matured across the ecosystem. +- The BIDS validator cannot stream content, so BIDS errors that require reading + a file (e.g., unreadable NIfTI headers) are suppressed for annexed files under + the ``stream`` and ``only-non-data`` policies. For NWB datasets, the primary + use case, nothing is lost: everything the BIDS validator needs is either + present (the non-annexed sidecar files, which are kept in git) or encoded in + the file and folder names of the annexed files themselves. + +.. _datalad-fuse: https://github.com/datalad/datalad-fuse + + Development Options ------------------- From 6db072412d33805c22c5150c0a8f62fbe456c0c6 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 6 Oct 2026 21:18:18 +0000 Subject: [PATCH 02/10] Document URL order and proxy behavior of stream policy Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Qn5WBSiQgoZoL4fytF6nEr --- docs/source/cmdline/validate.rst | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/docs/source/cmdline/validate.rst b/docs/source/cmdline/validate.rst index a789f6203..00b76eb97 100644 --- a/docs/source/cmdline/validate.rst +++ b/docs/source/cmdline/validate.rst @@ -113,7 +113,8 @@ a Dandiset:: --format=json_lines --output=000003.jsonl 000003 Each streamed file yields an ``INFO``-level ``DANDI.FILE_CONTENT_STREAMED`` -result naming the URL its content was read from; a file that cannot be streamed +result naming the first URL known for its content (for a Dandiset, the DANDI +Archive's API download URL, which redirects to S3); a file that cannot be streamed (not an annexed file, or no URL known to git-annex) yields a ``DANDI.FILE_CONTENT_MISSING`` error instead. @@ -121,6 +122,8 @@ Notes: - git-annex must be initialized in the clone, which ``datalad clone`` does (after ``git clone``, run ``git annex init``). +- datalad-fuse does not use the HTTP proxy environment variables + (``HTTPS_PROXY`` etc.), so the URLs must be reachable directly. - Some nwbinspector checks read data arrays (e.g., timestamps), so the amount of data streamed for a file depends on its content; it is nevertheless usually a small fraction of the file. From 0697ce06dde374bf41cc208c5e09b67433aa20c8 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 7 Oct 2026 01:40:01 +0000 Subject: [PATCH 03/10] TEMP: CI diagnostics for datalad job hang (to be reverted) Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Qn5WBSiQgoZoL4fytF6nEr --- .github/workflows/run-tests.yml | 12 +++++++++++- 1 file changed, 11 insertions(+), 1 deletion(-) diff --git a/.github/workflows/run-tests.yml b/.github/workflows/run-tests.yml index 7ee5117e8..fe521f3b4 100644 --- a/.github/workflows/run-tests.yml +++ b/.github/workflows/run-tests.yml @@ -152,7 +152,17 @@ jobs: # Start coverage before pytest, which loads dandi's pytest plugin (and # thus imports most of dandi) before pytest-cov would start it run: | - python -m coverage run -m pytest -s -v -m "not obolibrary" dandi + set +e + python -X faulthandler -m coverage run -m pytest -s -v -m "not obolibrary" dandi 2>&1 | tee "$RUNNER_TEMP/pytest.log" + status=${PIPESTATUS[0]} + if [ "$status" -ne 0 ]; then + # TEMPORARY diagnostics: surface the hang/crash tracebacks as an annotation + grep -n -B3 -A80 -E "Timeout|Fatal Python|Current thread|Thread 0x" "$RUNNER_TEMP/pytest.log" | head -400 > "$RUNNER_TEMP/diag.txt" + grep -E "^dandi/.*::" "$RUNNER_TEMP/pytest.log" | tail -5 >> "$RUNNER_TEMP/diag.txt" + echo "::error title=pytest diagnostics::$(sed -e 's/%/%25/g' -e 's/\r//g' "$RUNNER_TEMP/diag.txt" | sed ':a;N;$!ba;s/\n/%0A/g')" + exit "$status" + fi + set -e python -m coverage combine python -m coverage xml From 585d532756a01aea1e25d69d431839a583d919ab Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 7 Oct 2026 02:40:26 +0000 Subject: [PATCH 04/10] Revert "TEMP: CI diagnostics for datalad job hang (to be reverted)" This reverts commit d18843d73c449559c5c79b66c3bb7b770b98a584. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Qn5WBSiQgoZoL4fytF6nEr --- .github/workflows/run-tests.yml | 12 +----------- 1 file changed, 1 insertion(+), 11 deletions(-) diff --git a/.github/workflows/run-tests.yml b/.github/workflows/run-tests.yml index fe521f3b4..7ee5117e8 100644 --- a/.github/workflows/run-tests.yml +++ b/.github/workflows/run-tests.yml @@ -152,17 +152,7 @@ jobs: # Start coverage before pytest, which loads dandi's pytest plugin (and # thus imports most of dandi) before pytest-cov would start it run: | - set +e - python -X faulthandler -m coverage run -m pytest -s -v -m "not obolibrary" dandi 2>&1 | tee "$RUNNER_TEMP/pytest.log" - status=${PIPESTATUS[0]} - if [ "$status" -ne 0 ]; then - # TEMPORARY diagnostics: surface the hang/crash tracebacks as an annotation - grep -n -B3 -A80 -E "Timeout|Fatal Python|Current thread|Thread 0x" "$RUNNER_TEMP/pytest.log" | head -400 > "$RUNNER_TEMP/diag.txt" - grep -E "^dandi/.*::" "$RUNNER_TEMP/pytest.log" | tail -5 >> "$RUNNER_TEMP/diag.txt" - echo "::error title=pytest diagnostics::$(sed -e 's/%/%25/g' -e 's/\r//g' "$RUNNER_TEMP/diag.txt" | sed ':a;N;$!ba;s/\n/%0A/g')" - exit "$status" - fi - set -e + python -m coverage run -m pytest -s -v -m "not obolibrary" dandi python -m coverage combine python -m coverage xml From 266b2968bdabba86fdde169cad823d770ad7b841 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 12 Oct 2026 22:30:40 +0000 Subject: [PATCH 05/10] Cache NWB Inspector results, like those of PyNWB validation The results of inspecting an NWB file are cached in the validation cache, under the file (by stat() for a local file, or by its git-annex key for streamed content; see `annex_fingerprint`) and the version of nwbinspector, so that validating again does not inspect (and for streamed content, download) unchanged files again. A failure to read streamed content, which might be transient, is not cached. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Qn5WBSiQgoZoL4fytF6nEr --- dandi/files/bases.py | 144 +++++++++++++++++++++--------- dandi/validate/tests/test_core.py | 50 +++++++++++ 2 files changed, 153 insertions(+), 41 deletions(-) diff --git a/dandi/files/bases.py b/dandi/files/bases.py index 4bbc6dc5f..dc53f6bc8 100644 --- a/dandi/files/bases.py +++ b/dandi/files/bases.py @@ -2,16 +2,17 @@ from abc import ABC, abstractmethod from collections import deque -from collections.abc import Iterator +from collections.abc import Callable, Iterator from concurrent.futures import ThreadPoolExecutor, as_completed from dataclasses import dataclass from datetime import datetime +from functools import cache import os from pathlib import Path import re from threading import Lock import traceback -from typing import IO, Any, Generic +from typing import IO, Any, Generic, cast from xml.etree.ElementTree import fromstring import dandischema @@ -34,6 +35,7 @@ ) from dandi.metadata.core import get_default_metadata from dandi.misctypes import DUMMY_DANDI_ETAG, Digest, LocalReadableFile, P, Readable +from dandi.support.datalad_fuse import annex_fingerprint from dandi.utils import post_upload_size_check, pre_upload_size_check, yaml_load from dandi.validate._types import ( ORIGIN_INTERNAL_DANDI, @@ -636,56 +638,116 @@ def get_validation_errors( ) return errors - def _inspect_nwbfile(self, **kwargs: Any) -> Iterator[Any]: + def _inspect_nwbfile( + self, config: dict | None = None, importance_threshold: Any = None + ) -> list[Any]: """ - Yield the messages from inspecting the file with nwbinspector, reading + Return the messages from inspecting the file with nwbinspector, reading the content from `content_source` if it is set (in which case what `nwbinspector.inspect_nwbfile()` does for a local file is replicated here for the streamed content) """ # Avoid heavy import by importing within function: - from nwbinspector import ( - Importance, - InspectorMessage, - inspect_nwbfile, - inspect_nwbfile_object, - ) - - if self.content_source is None: - yield from inspect_nwbfile( - nwbfile_path=self.filepath, skip_validate=True, **kwargs - ) - return - - # Avoid heavy import by importing within function: - import h5py - from pynwb import NWBHDF5IO + from importlib.metadata import version - from dandi.pynwb_utils import open_readable + from nwbinspector import Importance, InspectorMessage + if importance_threshold is None: + importance_threshold = Importance.BEST_PRACTICE_SUGGESTION + source = self.content_source try: - with open_readable(self.content_source) as fp, h5py.File( - fp, "r" - ) as h5, NWBHDF5IO(file=h5, mode="r", load_namespaces=True) as io: - nwbfile = io.read() - for message in inspect_nwbfile_object(nwbfile_object=nwbfile, **kwargs): - if message is not None: - message.file_path = str(self.filepath) - yield message - except Exception as e: + return _cached_inspect_nwbfile()( + source if source is not None else str(self.filepath), + str(self.filepath), + version("nwbinspector"), + config, + importance_threshold, + ) + except _StreamedReadError as e: # Report the failure the same way inspect_nwbfile() does for a # local file that PyNWB cannot read - yield InspectorMessage( - message=traceback.format_exc(), - importance=Importance.ERROR, - check_function_name=( - f"During io.read(), an error occurred: " - f"{type(e).__module__}.{type(e).__name__}. " - "This indicates that PyNWB was unable to read the file. " - "See the traceback message for more details." - ), - file_path=str(self.filepath), - ) + exc = e.__cause__ + assert exc is not None + return [ + InspectorMessage( + message="".join(traceback.format_exception(exc)), + importance=Importance.ERROR, + check_function_name=( + f"During io.read(), an error occurred: " + f"{type(exc).__module__}.{type(exc).__name__}. " + "This indicates that PyNWB was unable to read the file. " + "See the traceback message for more details." + ), + file_path=str(self.filepath), + ) + ] + + +class _StreamedReadError(Exception): + """ + Failure to read or inspect streamed content, which is not cached as a + result of `_inspect_nwbfile()`, as it might be transient (e.g., a network + error) + """ + + +def _inspect_nwbfile( + source: str | Readable, + filepath: str, + nwbinspector_version: str, + config: dict | None, + importance_threshold: Any, +) -> list[Any]: + """ + Inspect the NWB file with nwbinspector, reading its content from ``source`` + (its path or a `Readable`), and return the messages, reporting + ``filepath`` as their file path + + The results are cached under ``source`` (see `annex_fingerprint`) and the + version of nwbinspector (which the caller passes so that it is part of the + key). + """ + # Avoid heavy import by importing within function: + from nwbinspector import inspect_nwbfile, inspect_nwbfile_object + + del nwbinspector_version + kwargs = {"config": config, "importance_threshold": importance_threshold} + if isinstance(source, str): + return list(inspect_nwbfile(nwbfile_path=source, skip_validate=True, **kwargs)) + + # Avoid heavy import by importing within function: + import h5py + from pynwb import NWBHDF5IO + + from dandi.pynwb_utils import open_readable + + messages = [] + try: + with open_readable(source) as fp, h5py.File(fp, "r") as h5, NWBHDF5IO( + file=h5, mode="r", load_namespaces=True + ) as io: + nwbfile = io.read() + for message in inspect_nwbfile_object(nwbfile_object=nwbfile, **kwargs): + if message is not None: + message.file_path = filepath + messages.append(message) + except Exception as e: + raise _StreamedReadError() from e + return messages + + +@cache +def _cached_inspect_nwbfile() -> Callable[..., list[Any]]: + """`_inspect_nwbfile()` memoized in the validation cache""" + # Avoid heavy import by importing within function: + from dandi.pynwb_utils import validate_cache + + return cast( + "Callable[..., list[Any]]", + validate_cache.memoize_path(custom_fingerprint=annex_fingerprint)( + _inspect_nwbfile + ), + ) class VideoAsset(LocalFileAsset): diff --git a/dandi/validate/tests/test_core.py b/dandi/validate/tests/test_core.py index ab3759542..39244178a 100644 --- a/dandi/validate/tests/test_core.py +++ b/dandi/validate/tests/test_core.py @@ -19,6 +19,7 @@ ) from ... import __version__ from ...consts import dandiset_metadata_file +from ...files import NWBAsset, dandi_file from ...pynwb_utils import validate as pynwb_validate from ...support.datalad_fuse import get_annexed_readable from ...tests.fixtures import ( @@ -420,6 +421,55 @@ def test_validate_stream_cached_by_key( assert server.ranged_requests == requests +def _streamed_nwb_asset(ds: Path) -> NWBAsset: + nwb = ds / "sub-001" / "sub-001.nwb" + df = dandi_file(nwb, ds) + assert isinstance(df, NWBAsset) + df.content_source = get_annexed_readable(nwb) + assert df.content_source is not None + return df + + +@pytest.mark.ai_generated +@pytest.mark.skipif( + os.environ.get("DANDI_CACHE") == "ignore", reason="the validation cache is disabled" +) +def test_validate_stream_inspection_cached_by_key( + streamable_dandiset: tuple[Path, RangeHTTPServer], +) -> None: + """NWB Inspector results of a streamed file are cached under its annex key.""" + ds, server = streamable_dandiset + results = _streamed_nwb_asset(ds).get_validation_errors() + inspector = [r for r in results if r.origin.validator == Validator.nwbinspector] + # simple3_nwb lacks a subject_id + assert inspector + requests = server.ranged_requests + # Served from the cache, without the content being read again + assert _streamed_nwb_asset(ds).get_validation_errors() == results + assert server.ranged_requests == requests + + +@pytest.mark.ai_generated +def test_validate_stream_inspection_failure_not_cached( + streamable_dandiset: tuple[Path, RangeHTTPServer], tmp_path: Path +) -> None: + """A failure to read streamed content for NWB Inspector is not cached.""" + ds, _ = streamable_dandiset + served = tmp_path / "served" / "content.nwb" + content = served.read_bytes() + served.unlink() + df = _streamed_nwb_asset(ds) + [msg] = df._inspect_nwbfile() + assert msg.importance.name == "ERROR" + assert msg.check_function_name.startswith("During io.read()") + assert msg.file_path == str(df.filepath) + served.write_bytes(content) + assert all( + m.check_function_name != msg.check_function_name + for m in _streamed_nwb_asset(ds)._inspect_nwbfile() + ) + + @pytest.mark.ai_generated def test_validate_stream_not_streamable(tmp_path: Path) -> None: """stream policy emits an error for a broken symlink that cannot be streamed.""" From 70dd2d5c1b16c84858687415d0b88dd29fe39631 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 12 Oct 2026 22:39:16 +0000 Subject: [PATCH 06/10] Add `dandi validate -J/--jobs` to validate NWB files in parallel Validating a streamed file mostly waits for its content to be downloaded, one request at a time (h5py serializes all reading of HDF5 files in a process), so validate NWB files (other than those of BIDS datasets, which are validated together) in a pool of processes, started with "spawn" since this process may run threads (e.g., fsspec's). The results are yielded in the same order as without the pool; the workers' log records are handled by the handlers of the main process. Streamed content is set up again in the worker, as its source cannot be sent to another process. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Qn5WBSiQgoZoL4fytF6nEr --- dandi/cli/cmd_validate.py | 20 ++- dandi/validate/_core.py | 256 +++++++++++++++++++++++------- dandi/validate/tests/test_core.py | 24 +++ 3 files changed, 240 insertions(+), 60 deletions(-) diff --git a/dandi/cli/cmd_validate.py b/dandi/cli/cmd_validate.py index ea18707c0..4bc7a9861 100644 --- a/dandi/cli/cmd_validate.py +++ b/dandi/cli/cmd_validate.py @@ -61,6 +61,7 @@ def _collect_results( devel_debug: bool, allow_any_path: bool, missing_file_content: MissingFileContent = MissingFileContent.error, + jobs: int = 1, ) -> list[ValidationResult]: """Run validation and collect all results into a list.""" # Avoid heavy import by importing within function: @@ -87,6 +88,7 @@ def _collect_results( devel_debug=devel_debug, allow_any_path=allow_any_path, missing_file_content=missing_file_content, + jobs=jobs, ) ) @@ -179,6 +181,16 @@ def _filter_results( ), default="error", ) +@click.option( + "-J", + "--jobs", + type=click.IntRange(min=1), + default=1, + show_default=True, + help="Number of NWB files to validate in parallel, in separate processes. " + "Useful especially with '--missing-file-content stream', for which " + "validating a file mostly waits for its content to be downloaded.", +) @click.option( "--load", help="Load validation results from JSONL file(s) instead of running validation.", @@ -201,6 +213,7 @@ def validate( summary: bool = False, max_per_group: int | None = None, missing_file_content: str = "error", + jobs: int = 1, load: tuple[str, ...] = (), schema: str | None = None, devel_debug: bool = False, @@ -244,7 +257,12 @@ def validate( else: mfc = MissingFileContent(missing_file_content) results = _collect_results( - paths, schema, devel_debug, allow_any_path, missing_file_content=mfc + paths, + schema, + devel_debug, + allow_any_path, + missing_file_content=mfc, + jobs=jobs, ) # Auto-save companion right after collection, before filtering — so # all results are preserved regardless of display filters. diff --git a/dandi/validate/_core.py b/dandi/validate/_core.py index e3c0e4f33..8e441dc81 100644 --- a/dandi/validate/_core.py +++ b/dandi/validate/_core.py @@ -9,11 +9,18 @@ from __future__ import annotations +from collections import deque from collections.abc import Iterator +from concurrent.futures import Future, ProcessPoolExecutor +from copy import copy from importlib.util import find_spec +import logging +from logging.handlers import QueueHandler, QueueListener +from multiprocessing import get_context import os from pathlib import Path import shutil +from typing import Any from ._types import ( ORIGIN_VALIDATION_DANDI_LAYOUT, @@ -27,7 +34,7 @@ Validator, ) from ..consts import dandiset_metadata_file -from ..files import DandiFile, LocalFileAsset, find_dandi_files +from ..files import DandiFile, LocalFileAsset, NWBAsset, find_dandi_files from ..support.datalad_fuse import AnnexedReadableFile, get_annexed_readable from ..utils import find_parent_directory_containing @@ -173,6 +180,7 @@ def validate( devel_debug: bool = False, allow_any_path: bool = False, missing_file_content: MissingFileContent = MissingFileContent.error, + jobs: int = 1, ) -> Iterator[ValidationResult]: """Validate content @@ -189,6 +197,11 @@ def validate( `dandi.support.datalad_fuse`) so that content-dependent validators run without the content being present locally. + jobs : int + Number of processes in which to validate NWB files (other than those of + BIDS datasets) in parallel; the results are yielded in the same order + as with 1 (the default), with which everything is validated in this + process. Yields ------ @@ -208,67 +221,192 @@ def validate( # The ids of the objects in `df_results` obtain through the `id()` built-in function df_result_ids: set[int] = set() - for p in paths: - p = os.path.abspath(p) - dandiset_path = find_parent_directory_containing(dandiset_metadata_file, p) - if dandiset_path is None: - yield ValidationResult( - id="DANDI.NO_DANDISET_FOUND", - origin=ORIGIN_VALIDATION_DANDI_LAYOUT, - severity=Severity.ERROR, - scope=Scope.DANDISET, - path=Path(p), - message="Path is not inside a Dandiset", - ) - for df in find_dandi_files( - p, dandiset_path=dandiset_path, allow_all=allow_any_path - ): - # Handle broken symlinks (missing file content) - is_broken = _is_broken_symlink(df.filepath) - if is_broken: - if missing_file_content == MissingFileContent.stream: - readable = _prepare_streaming(df) - # A file whose content cannot be streamed gets an error - # and is otherwise skipped - skip_file = readable is None + def unique(results: list[ValidationResult]) -> Iterator[ValidationResult]: + for r in results: + r_id = id(r) + if r_id not in df_result_ids: + df_results.append(r) + df_result_ids.add(r_id) + yield r + + # The results of the files, in order, either computed or being computed + # in `pool` + pending: deque[list[ValidationResult] | Future[list[ValidationResult]]] + pending = deque() + pool = _ValidationPool(jobs) if jobs > 1 else None + try: + for p in paths: + p = os.path.abspath(p) + dandiset_path = find_parent_directory_containing(dandiset_metadata_file, p) + if dandiset_path is None: + pending.append( + [ + ValidationResult( + id="DANDI.NO_DANDISET_FOUND", + origin=ORIGIN_VALIDATION_DANDI_LAYOUT, + severity=Severity.ERROR, + scope=Scope.DANDISET, + path=Path(p), + message="Path is not inside a Dandiset", + ) + ] + ) + for df in find_dandi_files( + p, dandiset_path=dandiset_path, allow_all=allow_any_path + ): + # Handle broken symlinks (missing file content) + is_broken = _is_broken_symlink(df.filepath) + if is_broken: + if missing_file_content == MissingFileContent.stream: + readable = _prepare_streaming(df) + # A file whose content cannot be streamed gets an error + # and is otherwise skipped + skip_file = readable is None + else: + readable = None + skip_file = missing_file_content in ( + MissingFileContent.skip, + MissingFileContent.error, + ) + pending.append( + [_handle_missing_content(df, missing_file_content, readable)] + ) + if skip_file: + continue + # only-non-data & stream: fall through but pass the policy + # to the validators + + file_policy = missing_file_content if is_broken else None + if pool is not None and type(df) is NWBAsset: + pending.append( + pool.submit(df, schema_version, devel_debug, file_policy) + ) else: - readable = None - skip_file = missing_file_content in ( - MissingFileContent.skip, - MissingFileContent.error, + pending.append( + _validate_file(df, schema_version, devel_debug, file_policy) ) - r = _handle_missing_content(df, missing_file_content, readable) - r_id = id(r) - if r_id not in df_result_ids: - df_results.append(r) - df_result_ids.add(r_id) - yield r - if skip_file: - continue - # only-non-data & stream: fall through but pass the policy to - # the validators - - for r in df.get_validation_errors( - schema_version=schema_version, - devel_debug=devel_debug, - missing_file_content=(missing_file_content if is_broken else None), - ): - # For broken-symlink files under only-non-data and stream, - # suppress BIDS errors that require reading file content (e.g. - # NIFTI_HEADER_UNREADABLE). The validator ran in full so - # real files still get those checks. - if ( - is_broken - and missing_file_content - in (MissingFileContent.only_non_data, MissingFileContent.stream) - and r.id in _BIDS_CONTENT_DEPENDENT_CODES + # Yield what is ready, or wait (for the oldest file) if enough + # files are being validated + while pending and ( + isinstance(pending[0], list) + or pending[0].done() + or (pool is not None and len(pending) > 2 * jobs) ): - continue - r_id = id(r) - if r_id not in df_result_ids: - df_results.append(r) - df_result_ids.add(r_id) - yield r + head = pending.popleft() + yield from unique(head if isinstance(head, list) else head.result()) + while pending: + head = pending.popleft() + yield from unique(head if isinstance(head, list) else head.result()) + finally: + if pool is not None: + pool.shutdown() + + +def _validate_file( + df: DandiFile, + schema_version: str | None, + devel_debug: bool, + missing_file_content: MissingFileContent | None, +) -> list[ValidationResult]: + """ + Return the results of validating ``df``; ``missing_file_content`` is the + policy for its missing content, if it is missing + """ + results = [] + for r in df.get_validation_errors( + schema_version=schema_version, + devel_debug=devel_debug, + missing_file_content=missing_file_content, + ): + # For broken-symlink files under only-non-data and stream, suppress + # BIDS errors that require reading file content (e.g. + # NIFTI_HEADER_UNREADABLE). The validator ran in full so real files + # still get those checks. + if ( + missing_file_content + in (MissingFileContent.only_non_data, MissingFileContent.stream) + and r.id in _BIDS_CONTENT_DEPENDENT_CODES + ): + continue + results.append(r) + return results + + +def _validate_file_in_worker( + df: LocalFileAsset, + schema_version: str | None, + devel_debug: bool, + missing_file_content: MissingFileContent | None, +) -> list[ValidationResult]: + """`_validate_file()` in a worker process of a `_ValidationPool`""" + if missing_file_content == MissingFileContent.stream: + # The streamed content is opened in this process + _prepare_streaming(df) + return _validate_file(df, schema_version, devel_debug, missing_file_content) + + +def _init_worker(log_queue: Any, levels: dict[str, int]) -> None: + """ + Set up a worker process of a `_ValidationPool`: send the log records to + the parent process, and ignore the warnings that it ignores + """ + # Avoid heavy import by importing within function: + from ..pynwb_utils import ignore_benign_pynwb_warnings + + root = logging.getLogger() + root.handlers = [QueueHandler(log_queue)] + for name, level in levels.items(): + logging.getLogger(name).setLevel(level) + ignore_benign_pynwb_warnings() + + +class _ValidationPool: + """ + Processes in which files are validated in parallel (h5py serializes all + reading of HDF5 files in a process), whose log records are handled by the + handlers of this process + """ + + def __init__(self, jobs: int) -> None: + # Not "fork": this process may have started threads (e.g., fsspec's + # for streaming content) + context = get_context("spawn") + self.log_queue = context.Queue() + self.listener = QueueListener( + self.log_queue, *logging.getLogger().handlers, respect_handler_level=True + ) + self.listener.start() + levels = {name: logging.getLogger(name or None).level for name in ("", "dandi")} + self.executor = ProcessPoolExecutor( + max_workers=jobs, + mp_context=context, + initializer=_init_worker, + initargs=(self.log_queue, levels), + ) + + def submit( + self, + df: LocalFileAsset, + schema_version: str | None, + devel_debug: bool, + missing_file_content: MissingFileContent | None, + ) -> Future[list[ValidationResult]]: + # The source of streamed content cannot be sent to another process; + # it is set up again there + df = copy(df) + df.content_source = None + return self.executor.submit( + _validate_file_in_worker, + df, + schema_version, + devel_debug, + missing_file_content, + ) + + def shutdown(self) -> None: + self.executor.shutdown(cancel_futures=True) + self.listener.stop() + self.log_queue.close() def _check_streaming_requirements() -> None: diff --git a/dandi/validate/tests/test_core.py b/dandi/validate/tests/test_core.py index 39244178a..4e54fd9e2 100644 --- a/dandi/validate/tests/test_core.py +++ b/dandi/validate/tests/test_core.py @@ -470,6 +470,30 @@ def test_validate_stream_inspection_failure_not_cached( ) +@pytest.mark.ai_generated +def test_validate_jobs( + streamable_dandiset: tuple[Path, RangeHTTPServer], simple3_nwb: Path +) -> None: + """Validating in parallel processes gives the same results, in order.""" + ds, server = streamable_dandiset + for sub in ("sub-002", "sub-003"): + (ds / sub).mkdir() + shutil.copy(simple3_nwb, ds / sub / f"{sub}.nwb") + # In parallel first, so that the streamed content is read in a worker + # process rather than the results taken from the cache + parallel = list( + validate(ds, missing_file_content=MissingFileContent.stream, jobs=2) + ) + assert server.ranged_requests > 0 + serial = list(validate(ds, missing_file_content=MissingFileContent.stream)) + assert any(r.origin.validator == Validator.nwbinspector for r in serial) + + def summary(results: list[ValidationResult]) -> list[tuple]: + return [(r.id, r.path, r.origin.validator, r.message) for r in results] + + assert summary(parallel) == summary(serial) + + @pytest.mark.ai_generated def test_validate_stream_not_streamable(tmp_path: Path) -> None: """stream policy emits an error for a broken symlink that cannot be streamed.""" From b82a255c8a5bc4a035ef8eec3d9b45855a991fc3 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 13 Oct 2026 17:04:58 +0000 Subject: [PATCH 07/10] Log the progress of `dandi validate`, one message per file validated Validating streamed content can take a long time per file, during which nothing was shown. Find all the files to validate first, log their number, and log for each file when it is done ("[3/120] sub-01/...nwb: 2 issues"), including for the files validated in parallel with --jobs, as they finish. The `dataset_description.json` files of BIDS datasets, to which the assets found refer only weakly, are kept alive while their assets are validated. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Qn5WBSiQgoZoL4fytF6nEr --- dandi/validate/_core.py | 183 ++++++++++++++++++++---------- dandi/validate/tests/test_core.py | 23 ++++ 2 files changed, 149 insertions(+), 57 deletions(-) diff --git a/dandi/validate/_core.py b/dandi/validate/_core.py index 8e441dc81..f02239c13 100644 --- a/dandi/validate/_core.py +++ b/dandi/validate/_core.py @@ -13,6 +13,7 @@ from collections.abc import Iterator from concurrent.futures import Future, ProcessPoolExecutor from copy import copy +from functools import partial from importlib.util import find_spec import logging from logging.handlers import QueueHandler, QueueListener @@ -20,6 +21,7 @@ import os from pathlib import Path import shutil +from threading import Lock from typing import Any from ._types import ( @@ -33,11 +35,21 @@ ValidationResult, Validator, ) +from .. import get_logger from ..consts import dandiset_metadata_file -from ..files import DandiFile, LocalFileAsset, NWBAsset, find_dandi_files +from ..files import ( + BIDSAsset, + DandiFile, + LocalAsset, + LocalFileAsset, + NWBAsset, + find_dandi_files, +) from ..support.datalad_fuse import AnnexedReadableFile, get_annexed_readable from ..utils import find_parent_directory_containing +lgr = get_logger() + BIDS_TO_DANDI = { "subject": "subject_id", "session": "session_id", @@ -233,67 +245,82 @@ def unique(results: list[ValidationResult]) -> Iterator[ValidationResult]: # in `pool` pending: deque[list[ValidationResult] | Future[list[ValidationResult]]] pending = deque() + # The files to validate (and the results for paths not in a Dandiset), all + # found first so that the progress can be reported as a count + items: list[DandiFile | ValidationResult] = [] + # The `dataset_description.json` files of BIDS datasets, which the + # `BIDSAsset` files found refer to only weakly + bids_descriptions = [] + for p in paths: + p = os.path.abspath(p) + dandiset_path = find_parent_directory_containing(dandiset_metadata_file, p) + if dandiset_path is None: + items.append( + ValidationResult( + id="DANDI.NO_DANDISET_FOUND", + origin=ORIGIN_VALIDATION_DANDI_LAYOUT, + severity=Severity.ERROR, + scope=Scope.DANDISET, + path=Path(p), + message="Path is not inside a Dandiset", + ) + ) + for df in find_dandi_files( + p, dandiset_path=dandiset_path, allow_all=allow_any_path + ): + if isinstance(df, BIDSAsset): + bids_descriptions.append(df.bids_dataset_description) + items.append(df) + progress = _Progress(sum(isinstance(i, DandiFile) for i in items)) + pool = _ValidationPool(jobs) if jobs > 1 else None try: - for p in paths: - p = os.path.abspath(p) - dandiset_path = find_parent_directory_containing(dandiset_metadata_file, p) - if dandiset_path is None: + for item in items: + if isinstance(item, ValidationResult): + pending.append([item]) + continue + df = item + # Handle broken symlinks (missing file content) + is_broken = _is_broken_symlink(df.filepath) + if is_broken: + if missing_file_content == MissingFileContent.stream: + readable = _prepare_streaming(df) + # A file whose content cannot be streamed gets an error + # and is otherwise skipped + skip_file = readable is None + else: + readable = None + skip_file = missing_file_content in ( + MissingFileContent.skip, + MissingFileContent.error, + ) pending.append( - [ - ValidationResult( - id="DANDI.NO_DANDISET_FOUND", - origin=ORIGIN_VALIDATION_DANDI_LAYOUT, - severity=Severity.ERROR, - scope=Scope.DANDISET, - path=Path(p), - message="Path is not inside a Dandiset", - ) - ] + [_handle_missing_content(df, missing_file_content, readable)] ) - for df in find_dandi_files( - p, dandiset_path=dandiset_path, allow_all=allow_any_path + if skip_file: + progress.done(df, None) + continue + # only-non-data & stream: fall through but pass the policy + # to the validators + + file_policy = missing_file_content if is_broken else None + if pool is not None and type(df) is NWBAsset: + future = pool.submit(df, schema_version, devel_debug, file_policy) + future.add_done_callback(partial(progress.future_done, df)) + pending.append(future) + else: + results = _validate_file(df, schema_version, devel_debug, file_policy) + progress.done(df, results) + pending.append(results) + # Yield what is ready, or wait (for the oldest file) if enough + # files are being validated + while pending and ( + isinstance(pending[0], list) + or pending[0].done() + or (pool is not None and len(pending) > 2 * jobs) ): - # Handle broken symlinks (missing file content) - is_broken = _is_broken_symlink(df.filepath) - if is_broken: - if missing_file_content == MissingFileContent.stream: - readable = _prepare_streaming(df) - # A file whose content cannot be streamed gets an error - # and is otherwise skipped - skip_file = readable is None - else: - readable = None - skip_file = missing_file_content in ( - MissingFileContent.skip, - MissingFileContent.error, - ) - pending.append( - [_handle_missing_content(df, missing_file_content, readable)] - ) - if skip_file: - continue - # only-non-data & stream: fall through but pass the policy - # to the validators - - file_policy = missing_file_content if is_broken else None - if pool is not None and type(df) is NWBAsset: - pending.append( - pool.submit(df, schema_version, devel_debug, file_policy) - ) - else: - pending.append( - _validate_file(df, schema_version, devel_debug, file_policy) - ) - # Yield what is ready, or wait (for the oldest file) if enough - # files are being validated - while pending and ( - isinstance(pending[0], list) - or pending[0].done() - or (pool is not None and len(pending) > 2 * jobs) - ): - head = pending.popleft() - yield from unique(head if isinstance(head, list) else head.result()) + head = pending.popleft() + yield from unique(head if isinstance(head, list) else head.result()) while pending: head = pending.popleft() yield from unique(head if isinstance(head, list) else head.result()) @@ -302,6 +329,48 @@ def unique(results: list[ValidationResult]) -> Iterator[ValidationResult]: pool.shutdown() +class _Progress: + """Report the progress of validating ``total`` files with log messages""" + + def __init__(self, total: int) -> None: + self.total = total + self.count = 0 + # Files may be done in the threads of a `_ValidationPool` + self.lock = Lock() + if total: + lgr.info("Validating %d file%s", total, "" if total == 1 else "s") + + def done( + self, + df: DandiFile, + results: list[ValidationResult] | None, + outcome: str = "skipped", + ) -> None: + """ + Report that validating ``df`` is done, with the given results, or + `None` if it was not validated, for the reason ``outcome`` + """ + with self.lock: + self.count += 1 + if results is None: + pass + elif results: + outcome = f"{len(results)} issue{'' if len(results) == 1 else 's'}" + else: + outcome = "OK" + path = df.path if isinstance(df, LocalAsset) else df.filepath + lgr.info("[%d/%d] %s: %s", self.count, self.total, path, outcome) + + def future_done( + self, df: DandiFile, future: Future[list[ValidationResult]] + ) -> None: + """Report that validating ``df`` in a `_ValidationPool` is done""" + if future.cancelled() or future.exception() is not None: + self.done(df, None, outcome="failed") + else: + self.done(df, future.result()) + + def _validate_file( df: DandiFile, schema_version: str | None, diff --git a/dandi/validate/tests/test_core.py b/dandi/validate/tests/test_core.py index 4e54fd9e2..23c296af5 100644 --- a/dandi/validate/tests/test_core.py +++ b/dandi/validate/tests/test_core.py @@ -1,4 +1,5 @@ import json +import logging import os from pathlib import Path import shutil @@ -470,6 +471,28 @@ def test_validate_stream_inspection_failure_not_cached( ) +@pytest.mark.ai_generated +@pytest.mark.parametrize("jobs", [1, 2]) +def test_validate_progress( + streamable_dandiset: tuple[Path, RangeHTTPServer], + simple3_nwb: Path, + caplog: pytest.LogCaptureFixture, + jobs: int, +) -> None: + """The progress of validation is logged, one message per file.""" + ds, _ = streamable_dandiset + (ds / "sub-002").mkdir() + shutil.copy(simple3_nwb, ds / "sub-002" / "sub-002.nwb") + caplog.set_level(logging.INFO, logger="dandi") + list(validate(ds, missing_file_content=MissingFileContent.stream, jobs=jobs)) + messages = [r.getMessage() for r in caplog.records] + messages = messages[messages.index("Validating 3 files") :] + done = [m for m in messages if m.startswith("[")] + assert sorted(m.split("] ")[0] for m in done) == ["[1/3", "[2/3", "[3/3"] + assert any(m.endswith("] sub-001/sub-001_video.mp4: OK") for m in done) + assert any("sub-002/sub-002.nwb: " in m and "issue" in m for m in done) + + @pytest.mark.ai_generated def test_validate_jobs( streamable_dandiset: tuple[Path, RangeHTTPServer], simple3_nwb: Path From 9eb666397bb7027b0b20d4e8fc23a036e49dfe73 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 15 Oct 2026 18:10:22 +0000 Subject: [PATCH 08/10] Stream with dandi.support.annex_stream (remfile) rather than datalad-fuse Follow the base branch, which streams annexed content with remfile from the URLs git-annex knows, without datalad-fuse: `--missing-file-content=stream` now requires remfile (`dandi[streaming]`) and git-annex, and never runs a `git annex` command in a clone in which git-annex is not initialized. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Qn5WBSiQgoZoL4fytF6nEr --- dandi/cli/cmd_validate.py | 8 ++++---- dandi/cli/tests/test_cmd_validate.py | 4 ++-- dandi/files/bases.py | 4 ++-- dandi/validate/_core.py | 12 ++++++------ dandi/validate/_types.py | 4 ++-- dandi/validate/tests/test_core.py | 19 ++++++++----------- docs/source/cmdline/validate.rst | 19 ++++++++++--------- 7 files changed, 34 insertions(+), 36 deletions(-) diff --git a/dandi/cli/cmd_validate.py b/dandi/cli/cmd_validate.py index 4bc7a9861..21a2513ac 100644 --- a/dandi/cli/cmd_validate.py +++ b/dandi/cli/cmd_validate.py @@ -172,10 +172,10 @@ def _filter_results( "in a datalad dataset without fetched data). 'error' (default) emits a " "concise error per file, 'skip' skips each such file with a warning, " "'only-non-data' skips content-dependent validators but still validates " - "path layout, 'stream' streams the content of annexed files with " - "datalad-fuse so that content-dependent validators run without the files " - "having to be downloaded (requires git-annex and `pip install " - "'dandi[datalad]'`).", + "path layout, 'stream' streams the content of annexed files from the URLs " + "git-annex knows for it, so that content-dependent validators run without " + "the files having to be downloaded (requires git-annex and `pip install " + "'dandi[streaming]'`).", type=click.Choice( ["error", "only-non-data", "skip", "stream"], case_sensitive=True ), diff --git a/dandi/cli/tests/test_cmd_validate.py b/dandi/cli/tests/test_cmd_validate.py index aef651740..10c426302 100644 --- a/dandi/cli/tests/test_cmd_validate.py +++ b/dandi/cli/tests/test_cmd_validate.py @@ -1003,12 +1003,12 @@ def test_validate_missing_file_content_no_broken_symlinks(tmp_path: Path) -> Non def test_validate_missing_file_content_stream( tmp_path: Path, simple3_nwb: Path, range_http_server: tuple[Any, str] ) -> None: - """--missing-file-content=stream validates annexed content with datalad-fuse.""" + """--missing-file-content=stream validates annexed content by streaming it.""" from ...tests.fixtures import make_git_annex_dandiset from ...tests.skip import skipif skipif.no_git_annex() - pytest.importorskip("datalad_fuse.fsspec") + pytest.importorskip("remfile") _, base_url = range_http_server shutil.copy(simple3_nwb, tmp_path / "served" / "content.nwb") ds = tmp_path / "ds" diff --git a/dandi/files/bases.py b/dandi/files/bases.py index dc53f6bc8..7c4f14c69 100644 --- a/dandi/files/bases.py +++ b/dandi/files/bases.py @@ -35,7 +35,7 @@ ) from dandi.metadata.core import get_default_metadata from dandi.misctypes import DUMMY_DANDI_ETAG, Digest, LocalReadableFile, P, Readable -from dandi.support.datalad_fuse import annex_fingerprint +from dandi.support.annex_stream import annex_fingerprint from dandi.utils import post_upload_size_check, pre_upload_size_check, yaml_load from dandi.validate._types import ( ORIGIN_INTERNAL_DANDI, @@ -326,7 +326,7 @@ class LocalFileAsset(LocalAsset): #: An alternative source for the asset's content, used instead of the file #: at `filepath` when set. This is used to stream the content of an #: annexed file (e.g., in a DataLad dataset) that is not present locally; - #: see `dandi.support.datalad_fuse`. + #: see `dandi.support.annex_stream`. content_source: Readable | None = None def _content(self) -> Path | Readable: diff --git a/dandi/validate/_core.py b/dandi/validate/_core.py index f02239c13..c8ceb6e1c 100644 --- a/dandi/validate/_core.py +++ b/dandi/validate/_core.py @@ -45,7 +45,7 @@ NWBAsset, find_dandi_files, ) -from ..support.datalad_fuse import AnnexedReadableFile, get_annexed_readable +from ..support.annex_stream import AnnexedReadableFile, get_annexed_readable from ..utils import find_parent_directory_containing lgr = get_logger() @@ -205,8 +205,8 @@ def validate( datalad dataset without fetched data). ``error`` emits a concise error, ``skip`` skips the file with a warning, ``only-non-data`` skips content-dependent validators but still validates path layout, and - ``stream`` streams the content of annexed files with datalad-fuse (see - `dandi.support.datalad_fuse`) so that + ``stream`` streams the content of annexed files (see + `dandi.support.annex_stream`) so that content-dependent validators run without the content being present locally. jobs : int @@ -481,10 +481,10 @@ def shutdown(self) -> None: def _check_streaming_requirements() -> None: """Raise an informative error if what is needed for streaming file content is not installed""" - if find_spec("datalad_fuse") is None: + if find_spec("remfile") is None: raise RuntimeError( - "Streaming file content requires datalad-fuse; install it with " - "`pip install 'dandi[datalad]'`" + "Streaming file content requires remfile; install it with " + "`pip install 'dandi[streaming]'`" ) if shutil.which("git-annex") is None: raise RuntimeError("Streaming file content requires git-annex") diff --git a/dandi/validate/_types.py b/dandi/validate/_types.py index 48b18b45a..56e2e1f62 100644 --- a/dandi/validate/_types.py +++ b/dandi/validate/_types.py @@ -27,8 +27,8 @@ class MissingFileContent(StrEnum): """Skip the file entirely; emit a WARNING noting that validation was skipped.""" stream = auto() - """Stream the content of annexed files with datalad-fuse (see - `dandi.support.datalad_fuse`) so that content-dependent validators run + """Stream the content of annexed files (see + `dandi.support.annex_stream`) so that content-dependent validators run without the content being present locally. An INFO result is emitted for each streamed file, and an ERROR for each file that cannot be streamed.""" diff --git a/dandi/validate/tests/test_core.py b/dandi/validate/tests/test_core.py index 23c296af5..6c484b338 100644 --- a/dandi/validate/tests/test_core.py +++ b/dandi/validate/tests/test_core.py @@ -22,12 +22,9 @@ from ...consts import dandiset_metadata_file from ...files import NWBAsset, dandi_file from ...pynwb_utils import validate as pynwb_validate -from ...support.datalad_fuse import get_annexed_readable -from ...tests.fixtures import ( - BIDS_TESTDATA_SELECTION, - RangeHTTPServer, - make_git_annex_dandiset, -) +from ...support.annex_stream import get_annexed_readable +from ...tests.fixtures import BIDS_TESTDATA_SELECTION, make_git_annex_dandiset +from ...tests.range_http_server import RangeHTTPServer from ...tests.skip import skipif @@ -338,7 +335,7 @@ def streamable_dandiset( its key is needed). """ skipif.no_git_annex() - pytest.importorskip("datalad_fuse.fsspec") + pytest.importorskip("remfile") server, base_url = range_http_server shutil.copy(simple3_nwb, tmp_path / "served" / "content.nwb") video = tmp_path / "video.mp4" @@ -371,7 +368,7 @@ def test_validate_stream( simple3_nwb: Path, tmp_path: Path, ) -> None: - """stream policy validates the content of annexed files with datalad-fuse.""" + """stream policy validates the content of annexed files by streaming it.""" ds, server = streamable_dandiset results = list(validate(ds, missing_file_content=MissingFileContent.stream)) @@ -520,7 +517,7 @@ def summary(results: list[ValidationResult]) -> list[tuple]: @pytest.mark.ai_generated def test_validate_stream_not_streamable(tmp_path: Path) -> None: """stream policy emits an error for a broken symlink that cannot be streamed.""" - pytest.importorskip("datalad_fuse.fsspec") + pytest.importorskip("remfile") skipif.no_git_annex() ds = _make_dandiset_with_broken_symlink(tmp_path) results = list(validate(ds, missing_file_content=MissingFileContent.stream)) @@ -557,9 +554,9 @@ def test_validate_stream_unreadable_url( @pytest.mark.ai_generated -def test_validate_stream_requires_datalad_fuse( +def test_validate_stream_requires_remfile( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: monkeypatch.setattr("dandi.validate._core.find_spec", lambda name: None) - with pytest.raises(RuntimeError, match=r"dandi\[datalad\]"): + with pytest.raises(RuntimeError, match=r"dandi\[streaming\]"): list(validate(tmp_path, missing_file_content=MissingFileContent.stream)) diff --git a/docs/source/cmdline/validate.rst b/docs/source/cmdline/validate.rst index 00b76eb97..7959d15a1 100644 --- a/docs/source/cmdline/validate.rst +++ b/docs/source/cmdline/validate.rst @@ -74,7 +74,8 @@ Options such file but still validate its path layout ``stream`` - Stream the content of each such file with datalad-fuse_ so that + Stream the content of each such file from the URLs git-annex knows + for it so that content-dependent validators run without the file having to be downloaded; see `Validating DataLad Dandisets Remotely`_ below. @@ -100,10 +101,10 @@ annexed files: symbolic links that remain broken until the content is fetched, which for some Dandisets would mean downloading terabytes of data. The ``stream`` policy of :option:`--missing-file-content` lets ``dandi validate`` run the content-dependent validators (pynwb and nwbinspector for NWB files) on -such files by streaming their content on demand with datalad-fuse_ (without -mounting anything), which asks git-annex where the content is, reading only -the parts of each file that the validators need. This requires git-annex and -``pip install "dandi[datalad]"``. +such files by streaming their content on demand: it asks git-annex where the +content is (``git annex whereis``) and reads only the parts of each file that +the validators need, with remfile_, which is optimized for HDF5-based files +such as NWB. This requires git-annex and ``pip install "dandi[streaming]"``. For example, to produce a JSON Lines record of *all* validation results for a Dandiset:: @@ -121,9 +122,9 @@ Archive's API download URL, which redirects to S3); a file that cannot be stream Notes: - git-annex must be initialized in the clone, which ``datalad clone`` does - (after ``git clone``, run ``git annex init``). -- datalad-fuse does not use the HTTP proxy environment variables - (``HTTPS_PROXY`` etc.), so the URLs must be reachable directly. + (after ``git clone``, run ``git annex init``): ``dandi validate`` does not + run any ``git annex`` command in a clone in which it is not, as that would + initialize it, modifying the clone. - Some nwbinspector checks read data arrays (e.g., timestamps), so the amount of data streamed for a file depends on its content; it is nevertheless usually a small fraction of the file. @@ -144,7 +145,7 @@ Notes: present (the non-annexed sidecar files, which are kept in git) or encoded in the file and folder names of the annexed files themselves. -.. _datalad-fuse: https://github.com/datalad/datalad-fuse +.. _remfile: https://github.com/flatironinstitute/remfile Development Options From 63ee181355132de073fed11ae1625cf9cab8adc1 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 15 Oct 2026 19:42:42 +0000 Subject: [PATCH 09/10] validate: stream with datalad-fuse (remfile backend); require `dandi[datalad]` Follow dandi.support.annex_stream streaming with datalad-fuse's adapter and its remfile backend: the stream policy now requires datalad-fuse with remfile (`pip install 'dandi[datalad]'`), checked with `can_stream()`, and the documentation says how the content is read. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Qn5WBSiQgoZoL4fytF6nEr --- dandi/cli/cmd_validate.py | 2 +- dandi/cli/tests/test_cmd_validate.py | 1 + dandi/validate/_core.py | 13 ++++++++----- dandi/validate/tests/test_core.py | 8 +++++--- docs/source/cmdline/validate.rst | 7 +++++-- 5 files changed, 20 insertions(+), 11 deletions(-) diff --git a/dandi/cli/cmd_validate.py b/dandi/cli/cmd_validate.py index 21a2513ac..c716925c8 100644 --- a/dandi/cli/cmd_validate.py +++ b/dandi/cli/cmd_validate.py @@ -175,7 +175,7 @@ def _filter_results( "path layout, 'stream' streams the content of annexed files from the URLs " "git-annex knows for it, so that content-dependent validators run without " "the files having to be downloaded (requires git-annex and `pip install " - "'dandi[streaming]'`).", + "'dandi[datalad]'`).", type=click.Choice( ["error", "only-non-data", "skip", "stream"], case_sensitive=True ), diff --git a/dandi/cli/tests/test_cmd_validate.py b/dandi/cli/tests/test_cmd_validate.py index 10c426302..e3f4186af 100644 --- a/dandi/cli/tests/test_cmd_validate.py +++ b/dandi/cli/tests/test_cmd_validate.py @@ -1008,6 +1008,7 @@ def test_validate_missing_file_content_stream( from ...tests.skip import skipif skipif.no_git_annex() + pytest.importorskip("datalad_fuse.adapter") pytest.importorskip("remfile") _, base_url = range_http_server shutil.copy(simple3_nwb, tmp_path / "served" / "content.nwb") diff --git a/dandi/validate/_core.py b/dandi/validate/_core.py index c8ceb6e1c..ec90b30f8 100644 --- a/dandi/validate/_core.py +++ b/dandi/validate/_core.py @@ -14,7 +14,6 @@ from concurrent.futures import Future, ProcessPoolExecutor from copy import copy from functools import partial -from importlib.util import find_spec import logging from logging.handlers import QueueHandler, QueueListener from multiprocessing import get_context @@ -45,7 +44,11 @@ NWBAsset, find_dandi_files, ) -from ..support.annex_stream import AnnexedReadableFile, get_annexed_readable +from ..support.annex_stream import ( + AnnexedReadableFile, + can_stream, + get_annexed_readable, +) from ..utils import find_parent_directory_containing lgr = get_logger() @@ -481,10 +484,10 @@ def shutdown(self) -> None: def _check_streaming_requirements() -> None: """Raise an informative error if what is needed for streaming file content is not installed""" - if find_spec("remfile") is None: + if not can_stream(): raise RuntimeError( - "Streaming file content requires remfile; install it with " - "`pip install 'dandi[streaming]'`" + "Streaming file content requires datalad-fuse with remfile; install " + "them with `pip install 'dandi[datalad]'`" ) if shutil.which("git-annex") is None: raise RuntimeError("Streaming file content requires git-annex") diff --git a/dandi/validate/tests/test_core.py b/dandi/validate/tests/test_core.py index 6c484b338..967142296 100644 --- a/dandi/validate/tests/test_core.py +++ b/dandi/validate/tests/test_core.py @@ -335,6 +335,7 @@ def streamable_dandiset( its key is needed). """ skipif.no_git_annex() + pytest.importorskip("datalad_fuse.adapter") pytest.importorskip("remfile") server, base_url = range_http_server shutil.copy(simple3_nwb, tmp_path / "served" / "content.nwb") @@ -517,6 +518,7 @@ def summary(results: list[ValidationResult]) -> list[tuple]: @pytest.mark.ai_generated def test_validate_stream_not_streamable(tmp_path: Path) -> None: """stream policy emits an error for a broken symlink that cannot be streamed.""" + pytest.importorskip("datalad_fuse.adapter") pytest.importorskip("remfile") skipif.no_git_annex() ds = _make_dandiset_with_broken_symlink(tmp_path) @@ -554,9 +556,9 @@ def test_validate_stream_unreadable_url( @pytest.mark.ai_generated -def test_validate_stream_requires_remfile( +def test_validate_stream_requires_datalad_fuse( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: - monkeypatch.setattr("dandi.validate._core.find_spec", lambda name: None) - with pytest.raises(RuntimeError, match=r"dandi\[streaming\]"): + monkeypatch.setattr("dandi.validate._core.can_stream", lambda: False) + with pytest.raises(RuntimeError, match=r"dandi\[datalad\]"): list(validate(tmp_path, missing_file_content=MissingFileContent.stream)) diff --git a/docs/source/cmdline/validate.rst b/docs/source/cmdline/validate.rst index 7959d15a1..541956959 100644 --- a/docs/source/cmdline/validate.rst +++ b/docs/source/cmdline/validate.rst @@ -103,8 +103,10 @@ which for some Dandisets would mean downloading terabytes of data. The run the content-dependent validators (pynwb and nwbinspector for NWB files) on such files by streaming their content on demand: it asks git-annex where the content is (``git annex whereis``) and reads only the parts of each file that -the validators need, with remfile_, which is optimized for HDF5-based files -such as NWB. This requires git-annex and ``pip install "dandi[streaming]"``. +the validators need. This is done with datalad-fuse_ (its adapter, without +mounting anything), which reads HDF5-based files such as NWB with remfile_, +optimized for their access pattern. This requires git-annex and ``pip install +"dandi[datalad]"``. For example, to produce a JSON Lines record of *all* validation results for a Dandiset:: @@ -145,6 +147,7 @@ Notes: present (the non-annexed sidecar files, which are kept in git) or encoded in the file and folder names of the annexed files themselves. +.. _datalad-fuse: https://github.com/datalad/datalad-fuse .. _remfile: https://github.com/flatironinstitute/remfile From 55dd75649b9ac64e1d1cbbe0558e0c6a93dabe88 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 16 Oct 2026 17:15:34 +0000 Subject: [PATCH 10/10] Ignore joblib's warning about slowly persisted arguments in tests joblib, which fscacher caches with, warns when saving the arguments of a cached call to its metadata.json takes over 0.5 s. The arguments of dandi's cached functions are small, so this happens only on a slow filesystem, as on a Windows CI runner in test_validate_simple1_no_subject, and the results are correct either way. With `filterwarnings = error`, the warning was raised inside the cached NWB Inspector call, and reported as the file's only validation error. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Qn5WBSiQgoZoL4fytF6nEr --- tox.ini | 3 +++ 1 file changed, 3 insertions(+) diff --git a/tox.ini b/tox.ini index 63dbff4f9..9974aab59 100644 --- a/tox.ini +++ b/tox.ini @@ -88,6 +88,9 @@ filterwarnings = ignore:.*unclosed.*:ResourceWarning:vcr # addressed in joblib 0.8.2-826-g05caf07 ignore:(ast.Num|Attribute n) is deprecated.*:DeprecationWarning:joblib + # Advisory warning of joblib (used by fscacher) when saving the arguments of + # a cached call takes over 0.5 s, e.g., on a slow filesystem + ignore:Persisting input arguments took:UserWarning:joblib [coverage:run] parallel = True