Replace click LazyFile with a custom type

src/check_jsonschema/cli/main_command.py

@@ -21,7 +21,7 @@
     SchemaLoaderBase,
 )
 from ..transforms import TRANSFORM_LIBRARY
-from .param_types import CommaDelimitedList, LazyBinaryReadFile, ValidatorClassName
+from .param_types import BinaryFileInputParam, CommaDelimitedList, ValidatorClassName
 from .parse_result import ParseResult, SchemaLoadingMode
 
 BUILTIN_SCHEMA_NAMES = [f"vendor.{k}" for k in SCHEMA_CATALOG.keys()] + [
@@ -119,11 +119,13 @@ def pretty_helptext_list(values: list[str] | tuple[str, ...]) -> str:
         "Instead of validating the instances against a schema, treat each file as a "
         "schema and validate them under their matching metaschemas."
     ),
+    default=False,
 )
 @click.option(
     "--no-cache",
     is_flag=True,
     help="Disable schema caching. Always download remote schemas.",
+    default=False,
 )
 @click.option(
     "--cache-filename", help="Deprecated. This option no longer has any effect."
@@ -191,6 +193,7 @@ def pretty_helptext_list(values: list[str] | tuple[str, ...]) -> str:
         "'--validator-class'"
     ),
     is_flag=True,
+    default=False,
 )
 @click.option(
     "--validator-class",
@@ -232,9 +235,7 @@ def pretty_helptext_list(values: list[str] | tuple[str, ...]) -> str:
     help="Reduce output verbosity",
     count=True,
 )
-@click.argument(
-    "instancefiles", required=True, nargs=-1, type=LazyBinaryReadFile("rb", lazy=True)
-)
+@click.argument("instancefiles", required=True, nargs=-1, type=BinaryFileInputParam())
 def main(
     *,
     schemafile: str | None,
@@ -255,7 +256,7 @@ def main(
     output_format: t.Literal["text", "json"],
     verbose: int,
     quiet: int,
-    instancefiles: tuple[t.IO[bytes], ...],
+    instancefiles: tuple[t.BinaryIO, ...],
 ) -> None:
     args = ParseResult()
 

src/check_jsonschema/cli/param_types.py

@@ -2,14 +2,14 @@
 
 import functools
 import importlib
-import os
 import re
-import stat
 import typing as t
 
 import click
 import jsonschema
-from click._compat import open_stream
+from click.shell_completion import CompletionItem
+
+from check_jsonschema.file_wrapper import BinaryFileInput
 
 C = t.TypeVar("C", bound=t.Callable[..., t.Any])
 
@@ -123,48 +123,20 @@ def convert(
         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,
-    ) -> None:
-        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 | os.PathLike[str] | t.IO[t.Any],
-        param: click.Parameter | None,
-        ctx: click.Context | None,
-    ) -> t.IO[bytes]:
-        if hasattr(value, "read") or hasattr(value, "write"):
-            return t.cast(t.IO[bytes], value)
-
-        value_: str | os.PathLike[str] = t.cast("str | os.PathLike[str]", value)
+class BinaryFileInputParam(click.ParamType[t.BinaryIO]):
+    def shell_complete(
+        self, ctx: click.Context, param: click.Parameter, incomplete: str
+    ) -> list[CompletionItem]:
+        return [CompletionItem(incomplete, type="file")]
 
-        lf = CustomLazyFile(value_, mode="rb")
+    def convert(
+        self, value: str, param: click.Parameter | None, ctx: click.Context | None
+    ) -> t.BinaryIO:
+        file = BinaryFileInput(value)
         if ctx is not None:
-            ctx.call_on_close(lf.close_intelligently)
-        return t.cast(t.IO[bytes], lf)
+            ctx.call_on_close(file.close)
+        file.check_can_open()
+        # type checkers won't recognize a BinaryFileInput as a BinaryIO unless we do a
+        # lot of extra work, because they don't follow that it uses `__getattr__` to
+        # proxy everything to the underlying stream
+        return file  # type: ignore[return-value]

src/check_jsonschema/cli/parse_result.py

@@ -23,7 +23,7 @@ def __init__(self) -> None:
         self.schema_mode: SchemaLoadingMode = SchemaLoadingMode.filepath
         self.schema_path: str | None = None
         self.base_uri: str | None = None
-        self.instancefiles: tuple[t.IO[bytes], ...] = ()
+        self.instancefiles: tuple[t.BinaryIO, ...] = ()
         # cache controls
         self.disable_cache: bool = False
         self.cache_filename: str | None = None

src/check_jsonschema/file_wrapper.py

@@ -0,0 +1,84 @@
+"""
+A file-like object specifically for use within check-jsonschema.
+
+Inspired heavily by `click`'s internal `LazyFile` object.
+
+Features:
+
+- fixed mode ("rb")
+- a "check_can_open()" function which will try to open and close the file, but only
+  if it looks like a regular file
+- passthrough of most attributes to the underlying file object
+- automatically grabs binary stdin when the input name is "-"
+- `close()` does nothing on stdin
+"""
+
+import io
+import os
+import stat
+import sys
+import types
+import typing as t
+
+if sys.version_info >= (3, 11):
+    from typing import Self
+else:
+    from typing_extensions import Self
+
+
+class BinaryFileInput:
+    def __init__(self, name: str) -> None:
+        self.name = name
+
+        self._stream: t.BinaryIO | None = None
+        self._is_stdin: bool = False
+        if self.name == "-":
+            self._stream = sys.stdin.buffer
+            self._is_stdin = True
+
+    def check_can_open(self) -> None:
+        if self._is_stdin:
+            return
+
+        if not stat.S_ISFIFO(os.stat(self.name).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.
+            self.open()
+            self.close()
+
+    def open(self) -> t.BinaryIO:
+        if self._stream is None:
+            self._stream = open(self.name, "rb")
+        return self._stream
+
+    def close(self) -> None:
+        if self._is_stdin:
+            return
+        if self._stream is not None:
+            self._stream.close()
+            self._stream = None
+
+    def __getattr__(self, name: str) -> t.Any:
+        return getattr(self.open(), name)
+
+    def __repr__(self) -> str:
+        return "BinaryFileInput(filename={self.filename!r})"
+
+    def __enter__(self) -> Self:
+        return self
+
+    def __exit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc_value: BaseException | None,
+        tb: types.TracebackType | None,
+    ) -> None:
+        self.close()
+
+    def __iter__(self) -> t.Iterator[bytes]:
+        return iter(self.open())
+
+
+# register as an IOBase implementation, since pyjson5 relies on an isinstance() check
+io.IOBase.register(BinaryFileInput)

src/check_jsonschema/instance_loader.py

@@ -3,16 +3,14 @@
 import io
 import typing as t
 
-from check_jsonschema.cli.param_types import CustomLazyFile
-
 from .parsers import ParseError, ParserSet
 from .transforms import Transform
 
 
 class InstanceLoader:
     def __init__(
         self,
-        files: t.Sequence[t.IO[bytes] | CustomLazyFile],
+        files: t.Sequence[t.BinaryIO],
         default_filetype: str = "json",
         force_filetype: str | None = None,
         data_transform: Transform | None = None,
@@ -41,14 +39,9 @@ def iter_files(self) -> t.Iterator[tuple[str, ParseError | t.Any]]:
                 raise ValueError(f"File {file} has no name attribute")
 
             try:
-                if isinstance(file, CustomLazyFile):
-                    stream: t.IO[bytes] = t.cast(t.IO[bytes], file.open())
-                else:
-                    stream = file
-
                 try:
                     data: t.Any = self._parsers.parse_data_with_path(
-                        stream, name, self._default_filetype, self._force_filetype
+                        file, name, self._default_filetype, self._force_filetype
                     )
                 except ParseError as err:
                     data = err

tests/unit/cli/test_parse.py

@@ -7,6 +7,7 @@
 
 from check_jsonschema import main as cli_main
 from check_jsonschema.cli.parse_result import ParseResult, SchemaLoadingMode
+from check_jsonschema.file_wrapper import BinaryFileInput
 
 
 class BoxedContext:
@@ -81,7 +82,7 @@ def test_schemafile_and_instancefile(
     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, click.utils.LazyFile)
+        assert isinstance(f, BinaryFileInput)
     assert tuple(f.name for f in mock_parse_result.instancefiles) == ("foo.json",)