Apply lazy file reading for instancefiles

Lazy reads are slightly customized to better handle named pipes.

The lazy file reads are combined with an early close immediately after
the instance loader gets data from a file. As a result,
check-jsonschema should only keep one instance file open at a time.

Author: sirosen

src/check_jsonschema/cli/main_command.py

@@ -20,7 +20,7 @@
     SchemaLoaderBase,
 )
 from ..transforms import TRANSFORM_LIBRARY
-from .param_types import CommaDelimitedList, ValidatorClassName
+from .param_types import CommaDelimitedList, LazyBinaryReadFile, ValidatorClassName
 from .parse_result import ParseResult, SchemaLoadingMode
 
 BUILTIN_SCHEMA_NAMES = [f"vendor.{k}" for k in SCHEMA_CATALOG.keys()] + [
@@ -220,7 +220,9 @@ def pretty_helptext_list(values: list[str] | tuple[str, ...]) -> str:
     help="Reduce output verbosity",
     count=True,
 )
-@click.argument("instancefiles", required=True, nargs=-1, type=click.File("rb"))
+@click.argument(
+    "instancefiles", required=True, nargs=-1, type=LazyBinaryReadFile("rb", lazy=True)
+)
 def main(
     *,
     schemafile: str | None,

src/check_jsonschema/cli/param_types.py

@@ -1,11 +1,14 @@
 from __future__ import annotations
 
 import importlib
+import os
 import re
+import stat
 import typing as t
 
 import click
 import jsonschema
+from click._compat import open_stream
 
 
 class CommaDelimitedList(click.ParamType):
@@ -104,3 +107,45 @@ def convert(
             self.fail(f"'{classname}' in '{pkg}' is not a class", param, ctx)
 
         return t.cast(t.Type[jsonschema.protocols.Validator], result)
+
+
+class _CustomLazyFile(click.utils.LazyFile):
+    def __init__(
+        self,
+        filename: str | os.PathLike[str],
+        mode: str = "r",
+        encoding: str | None = None,
+        errors: str | None = "strict",
+        atomic: bool = False,
+    ):
+        self.name: str = os.fspath(filename)
+        self.mode = mode
+        self.encoding = encoding
+        self.errors = errors
+        self.atomic = atomic
+        self._f: t.IO[t.Any] | None
+        self.should_close: bool
+
+        if self.name == "-":
+            self._f, self.should_close = open_stream(filename, mode, encoding, errors)
+        else:
+            if "r" in mode and not stat.S_ISFIFO(os.stat(filename).st_mode):
+                # Open and close the file in case we're opening it for
+                # reading so that we can catch at least some errors in
+                # some cases early.
+                open(filename, mode).close()
+            self._f = None
+            self.should_close = True
+
+
+class LazyBinaryReadFile(click.File):
+    def convert(
+        self,
+        value: str,
+        param: click.Parameter | None,
+        ctx: click.Context | None,
+    ) -> t.IO[bytes]:
+        lf = _CustomLazyFile(value, mode="rb")
+        if ctx is not None:
+            ctx.call_on_close(lf.close_intelligently)
+        return t.cast(t.IO[bytes], lf)

src/check_jsonschema/instance_loader.py

@@ -43,4 +43,6 @@ def iter_files(self) -> t.Iterator[tuple[str, ParseError | t.Any]]:
                 data = err
             else:
                 data = self._data_transform(data)
+
+            file.close()
             yield (name, data)

tests/acceptance/test_special_filetypes.py

@@ -1,7 +1,7 @@
+import multiprocessing
 import os
 import platform
 import sys
-import threading
 
 import pytest
 import responses
@@ -33,6 +33,16 @@ def test_schema_and_instance_in_memfds(run_line_simple):
         os.close(instancefd)
 
 
+# helper (in global scope) for multiprocessing "spawn" to be able to use to launch
+# background writers
+def _fifo_write(path, data):
+    fd = os.open(path, os.O_WRONLY)
+    try:
+        os.write(fd, data)
+    finally:
+        os.close(fd)
+
+
 @pytest.mark.skipif(os.name != "posix", reason="test requires mkfifo")
 @pytest.mark.parametrize("check_succeeds", (True, False))
 def test_schema_and_instance_in_fifos(tmp_path, run_line, check_succeeds):
@@ -45,25 +55,17 @@ def test_schema_and_instance_in_fifos(tmp_path, run_line, check_succeeds):
     os.mkfifo(schema_path)
     os.mkfifo(instance_path)
 
-    # execute FIFO writes as blocking writes in background threads
-    # nonblocking writes fail file existence if there's no reader, so using a FIFO
-    # requires some level of concurrency
-    def fifo_write(path, data):
-        fd = os.open(path, os.O_WRONLY)
-        try:
-            os.write(fd, data)
-        finally:
-            os.close(fd)
-
-    schema_thread = threading.Thread(
-        target=fifo_write, args=[schema_path, b'{"type": "integer"}']
+    spawn_ctx = multiprocessing.get_context("spawn")
+
+    schema_proc = spawn_ctx.Process(
+        target=_fifo_write, args=(schema_path, b'{"type": "integer"}')
     )
-    schema_thread.start()
+    schema_proc.start()
     instance_data = b"42" if check_succeeds else b'"foo"'
-    instance_thread = threading.Thread(
-        target=fifo_write, args=[instance_path, instance_data]
+    instance_proc = spawn_ctx.Process(
+        target=_fifo_write, args=(instance_path, instance_data)
     )
-    instance_thread.start()
+    instance_proc.start()
 
     try:
         result = run_line(
@@ -74,8 +76,8 @@ def fifo_write(path, data):
         else:
             assert result.exit_code == 1
     finally:
-        schema_thread.join(timeout=0.1)
-        instance_thread.join(timeout=0.1)
+        schema_proc.terminate()
+        instance_proc.terminate()
 
 
 @pytest.mark.parametrize("check_passes", (True, False))

tests/unit/test_cli_parse.py

@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-import io
 from unittest import mock
 
 import click
@@ -86,7 +85,7 @@ def test_schemafile_and_instancefile(runner, mock_parse_result, in_tmp_dir, tmp_
     assert mock_parse_result.schema_path == "schema.json"
     assert isinstance(mock_parse_result.instancefiles, tuple)
     for f in mock_parse_result.instancefiles:
-        assert isinstance(f, (io.BytesIO, io.BufferedReader))
+        assert isinstance(f, click.utils.LazyFile)
     assert tuple(f.name for f in mock_parse_result.instancefiles) == ("foo.json",)