optimize validating multiple files by allowing to call the cli only once and to reuse the schemas without reloading them

doclang/_schemas.py

@@ -10,6 +10,7 @@
 
 import re
 from dataclasses import dataclass
+from functools import cache
 from pathlib import Path
 
 from lxml import etree
@@ -63,6 +64,11 @@ def _version_key(version: str) -> tuple[int, int]:
 
 def _schema_version(xsd: Path) -> str:
     """The ``MAJOR.MINOR`` spec version of an XSD, from its ``xs:schema`` version attribute."""
+    return _schema_version_cached(xsd, xsd.stat().st_mtime_ns)
+
+
+@cache
+def _schema_version_cached(xsd: Path, mtime_ns: int) -> str:
     match = _SCHEMA_VERSION_RE.search(xsd.read_text(encoding="utf-8"))
     if match is None:
         raise FileNotFoundError(f"No schema version found in {xsd}")
@@ -128,12 +134,17 @@ def _head_description() -> str:
     return "unreleased development version"
 
 
-def _enumerated_versions(xsd: Path) -> set[str]:
-    """Versions the XSD enumerates for the root ``version`` attribute (empty if only format-constrained)."""
+@cache
+def _enumerated_versions_cached(xsd: str, mtime_ns: int) -> frozenset[str]:
     parser = etree.XMLParser(resolve_entities=False, no_network=True)
-    tree = etree.parse(str(xsd), parser)
+    tree = etree.parse(xsd, parser)
     values = tree.xpath("//xs:attribute[@name='version']//xs:enumeration/@value", namespaces={"xs": _XS_NS})
-    return {str(v) for v in values}  # type: ignore[union-attr]
+    return frozenset(str(v) for v in values)  # type: ignore[union-attr]
+
+
+def _enumerated_versions(xsd: Path) -> set[str]:
+    """Versions the XSD enumerates for the root ``version`` attribute (empty if only format-constrained)."""
+    return set(_enumerated_versions_cached(str(xsd), xsd.stat().st_mtime_ns))
 
 
 def _declared_version(xml_file: str | Path) -> str | None:

doclang/backends/saxonche.py

@@ -8,6 +8,7 @@
 
 import tempfile
 from pathlib import Path
+from typing import Any
 
 from lxml import etree
 
@@ -123,7 +124,55 @@ def _transpile_schematron_to_xslt(sch_file: Path, *, verbose: bool = False) -> s
 
 
 class SaxoncheValidator:
-    """Schematron validator backed by Saxon/C (saxonche)."""
+    """Schematron validator backed by Saxon/C (saxonche).
+
+    One instance keeps a single Saxon processor and the compiled XSLT of every Schematron file it has
+    seen, so reusing it for several documents transpiles and compiles each schema only once. Call
+    :meth:`close` (or use the instance as a context manager) to release the processor when done.
+    """
+
+    def __init__(self) -> None:
+        self._proc: Any = None
+        self._xslt_proc: Any = None
+        self._executables: dict[tuple[str, int], Any] = {}
+
+    def close(self) -> None:
+        """Release the Saxon processor and the compiled schemas."""
+        self._executables.clear()
+        self._xslt_proc = None
+        if self._proc is not None:
+            self._proc.__exit__(None, None, None)
+            self._proc = None
+
+    def __enter__(self) -> SaxoncheValidator:
+        return self
+
+    def __exit__(self, *exc_info: object) -> None:
+        self.close()
+
+    def _executable(self, schema_path: Path, *, verbose: bool) -> Any:
+        """The compiled XSLT for a Schematron file, built on first use."""
+        _require_saxonche_backend()
+        from saxonche import PySaxonProcessor
+
+        if self._proc is None:
+            self._proc = PySaxonProcessor(license=False).__enter__()
+            self._xslt_proc = self._proc.new_xslt30_processor()
+            if verbose:
+                print(f"Using XSLT processor version: {self._proc.version}")
+
+        # Keyed on the modification time so an edited schema is not served stale.
+        key = (str(schema_path), schema_path.stat().st_mtime_ns)
+        executable = self._executables.get(key)
+        if executable is None:
+            if verbose:
+                print("Transpiling Schematron to XSLT 3.0...")
+            xslt_text = _transpile_schematron_to_xslt(schema_path, verbose=verbose)
+            if verbose:
+                print("Compiling generated XSLT...")
+            executable = self._xslt_proc.compile_stylesheet(stylesheet_text=xslt_text)
+            self._executables[key] = executable
+        return executable
 
     def validate(
         self,
@@ -133,9 +182,6 @@ def validate(
         allow_empty_namespace: bool = False,
         verbose: bool = False,
     ) -> list[SchematronViolation]:
-        _require_saxonche_backend()
-        from saxonche import PySaxonProcessor
-
         if verbose:
             print(f"Using Schematron file: {schema_path}")
 
@@ -145,35 +191,20 @@ def validate(
         if allow_empty_namespace:
             xml_doc = _ensure_namespace(xml_doc)
 
+        xslt_executable = self._executable(schema_path, verbose=verbose)
+
         with tempfile.NamedTemporaryFile(mode="wb", suffix=".xml", delete=True) as tmp:
             # Never re-emit a DOCTYPE: Saxon would otherwise expand entities.
             _write_xml_without_dtd(xml_doc, tmp)
             tmp.flush()
-            tmp_xml_path = tmp.name
-
-            with PySaxonProcessor(license=False) as proc:
-                if verbose:
-                    print(f"Using XSLT processor version: {proc.version}")
-
-                xslt_proc = proc.new_xslt30_processor()
-
-                if verbose:
-                    print("Transpiling Schematron to XSLT 3.0...")
-
-                xslt_text = _transpile_schematron_to_xslt(schema_path, verbose=verbose)
-
-                if verbose:
-                    print("Compiling generated XSLT...")
-
-                xslt_executable = xslt_proc.compile_stylesheet(stylesheet_text=xslt_text)
 
-                if verbose:
-                    print("Executing Schematron validation...")
+            if verbose:
+                print("Executing Schematron validation...")
 
-                result = xslt_executable.transform_to_string(source_file=tmp_xml_path)
+            result = xslt_executable.transform_to_string(source_file=tmp.name)
 
-                if not result:
-                    return []
+            if not result:
+                return []
 
-                result_doc = etree.fromstring(result.encode("utf-8"))
-                return _svrl_failed_asserts_to_violations(result_doc)
+            result_doc = etree.fromstring(result.encode("utf-8"))
+            return _svrl_failed_asserts_to_violations(result_doc)

doclang/cli.py

@@ -18,7 +18,7 @@
 from doclang.migration import migrate as migrate_document
 from doclang.packaging import PackagingError
 from doclang.packaging import pack as pack_document
-from doclang.schematron import SchematronBackendNotFound
+from doclang.schematron import SchematronBackendNotFound, SchematronValidator, _default_schematron_validator
 from doclang.utils import _VERSION
 from doclang.validation import ValidationError, VersionNote
 from doclang.validation import validate as validate_document
@@ -53,7 +53,18 @@ class OutputFormat(str, Enum):
 )
 def validate(
     ctx: typer.Context,
-    xml_file: Path = typer.Argument(..., help="XML file or .dclx archive to validate", exists=True),
+    xml_file: Optional[Path] = typer.Argument(None, help="XML file or .dclx archive to validate", exists=True),
+    filelist: Optional[Path] = typer.Option(
+        None,
+        "--filelist",
+        help=(
+            "File containing a list of files to validate, one path per line, or '-' to read the list from "
+            "standard input (blank lines and lines starting with '#' are ignored)"
+        ),
+        exists=True,
+        dir_okay=False,
+        allow_dash=True,
+    ),
     xsd_only: bool = typer.Option(False, "--xsd-only", help="Validate XSD only"),
     schematron_only: bool = typer.Option(False, "--schematron-only", help="Validate Schematron only"),
     allow_empty_namespace: bool = typer.Option(
@@ -101,7 +112,95 @@ def validate(
         doclang validate document.xml --spec-version 0.7
         doclang validate document.xml --spec-version 0.7 --strict-xsd-version-attr
         doclang validate document.xml --format json
+        doclang validate --filelist files.txt
+        find . -name "*.dclg" | doclang validate --filelist -
     """
+    files: list[Path] = []
+    if xml_file is not None:
+        files.append(xml_file)
+    if filelist is not None:
+        try:
+            if str(filelist) == "-":
+                lines = typer.get_text_stream("stdin").read().splitlines()
+            else:
+                lines = filelist.read_text(encoding="utf-8").splitlines()
+        except (OSError, UnicodeDecodeError) as exc:
+            typer.echo(f"Error: cannot read file list {filelist}: {exc}", err=True)
+            raise typer.Exit(1)
+        files.extend(Path(line.strip()) for line in lines if line.strip() and not line.strip().startswith("#"))
+    if not files:
+        typer.echo("Error: provide an XML_FILE argument or --filelist", err=True)
+        raise typer.Exit(2)
+
+    # One validator for all files: it keeps the compiled Schematron, so each schema is processed once.
+    schematron: Optional[SchematronValidator] = None
+    if not xsd_only:
+        try:
+            schematron = _default_schematron_validator()
+        except SchematronBackendNotFound:
+            pass  # reported for each file by validate_document
+        else:
+            ctx.call_on_close(schematron.close)  # type: ignore[attr-defined]
+
+    seen_schemas: set[Path] = set()
+    failed = 0
+    for path in files:
+        if not path.exists():
+            typer.echo(f"Error: file not found: {path}", err=True)
+            failed += 1
+            continue
+        try:
+            _validate_one(
+                ctx,
+                path,
+                seen_schemas=seen_schemas,
+                schematron=schematron,
+                xsd_only=xsd_only,
+                schematron_only=schematron_only,
+                allow_empty_namespace=allow_empty_namespace,
+                spec_version=spec_version,
+                strict_xsd_version_attr=strict_xsd_version_attr,
+                strict_declared_version=strict_declared_version,
+                quiet=quiet,
+                verbose=verbose,
+                format=format,
+            )
+        except typer.Exit as exc:
+            if exc.exit_code:
+                failed += 1
+
+    if len(files) > 1 and not quiet and format == OutputFormat.text:
+        typer.echo(f"Validated {len(files)} files: {len(files) - failed} passed, {failed} failed")
+    if failed:
+        raise typer.Exit(1)
+
+
+def _schema_line(schema: Optional[Path], seen_schemas: set[Path]) -> str:
+    """Verbose line for a schema: processed on first use, then reused from the in-memory cache."""
+    if schema is None:
+        return "Schema: n/a"
+    status = "reused, already processed" if schema in seen_schemas else "processing"
+    seen_schemas.add(schema)
+    return f"Schema: {schema} ({status})"
+
+
+def _validate_one(
+    ctx: typer.Context,
+    xml_file: Path,
+    *,
+    seen_schemas: set[Path],
+    schematron: Optional[SchematronValidator],
+    xsd_only: bool,
+    schematron_only: bool,
+    allow_empty_namespace: bool,
+    spec_version: Optional[str],
+    strict_xsd_version_attr: bool,
+    strict_declared_version: bool,
+    quiet: bool,
+    verbose: bool,
+    format: OutputFormat,
+) -> None:
+    """Validate a single file; raises typer.Exit(1) on failure."""
     try:
         document = ctx.with_resource(_document_file(xml_file))
     except PackagingError as exc:
@@ -133,6 +232,7 @@ def validate(
                 validate_document(
                     document,
                     allow_empty_namespace=allow_empty_namespace,
+                    schematron=schematron,
                     xsd_only=xsd_only,
                     schematron_only=schematron_only,
                     spec_version=spec_version,
@@ -159,7 +259,7 @@ def validate(
             if not schematron_only:
                 if verbose:
                     typer.echo("XSD Validation")
-                    typer.echo(f"Schema: {bundled_xsd or 'n/a'}")
+                    typer.echo(_schema_line(bundled_xsd, seen_schemas))
                 if not exc.xsd_errors:
                     typer.echo("XSD validation passed")
                 else:
@@ -173,7 +273,7 @@ def validate(
             if not xsd_only:
                 if verbose:
                     typer.echo("Schematron Validation")
-                    typer.echo(f"Schema: {bundled_sch or 'n/a'}")
+                    typer.echo(_schema_line(bundled_sch, seen_schemas))
                 if not exc.schematron_errors:
                     typer.echo("Schematron validation passed")
                 else:
@@ -194,13 +294,13 @@ def validate(
         if not schematron_only:
             if verbose:
                 typer.echo("XSD Validation")
-                typer.echo(f"Schema: {bundled_xsd or 'n/a'}")
+                typer.echo(_schema_line(bundled_xsd, seen_schemas))
             typer.echo("XSD validation passed")
 
         if not xsd_only:
             if verbose:
                 typer.echo("Schematron Validation")
-                typer.echo(f"Schema: {bundled_sch or 'n/a'}")
+                typer.echo(_schema_line(bundled_sch, seen_schemas))
             typer.echo("Schematron validation passed")
 
     if format == OutputFormat.json:

doclang/xsd_validation.py

@@ -5,6 +5,7 @@
 """
 
 import re
+from functools import cache
 from pathlib import Path
 from typing import Any, Union
 
@@ -14,6 +15,14 @@
 from doclang.utils import _ensure_namespace, _parse_doclang_document, _parse_xml
 
 
+@cache
+def _compiled_schema(xsd_file: str, mtime_ns: int) -> etree.XMLSchema:
+    """Parse and compile an XSD once per file (``mtime_ns`` invalidates the entry when the file changes)."""
+    with open(xsd_file, "rb") as f:
+        # Bundled/trusted schema: hardened parse only (no DocLang DTD policy).
+        return etree.XMLSchema(_parse_xml(f))
+
+
 def _validate_xsd_at(
     xml_file: Union[str, Path],
     xsd_file: Union[str, Path],
@@ -37,10 +46,7 @@ def _validate_xsd_at(
         Validation errors as a list of dicts with 'line' and 'message' keys
     """
     try:
-        with open(xsd_file, "rb") as f:
-            # Bundled/trusted schema: hardened parse only (no DocLang DTD policy).
-            schema_doc = _parse_xml(f)
-            schema = etree.XMLSchema(schema_doc)
+        schema = _compiled_schema(str(xsd_file), Path(xsd_file).stat().st_mtime_ns)
 
         with open(xml_file, "rb") as f:
             xml_doc = _parse_doclang_document(f)