Add support for passing --schemafile '-' for stdin

Author: sirosen

src/check_jsonschema/cli/main_command.py

@@ -91,8 +91,10 @@ def pretty_helptext_list(values: list[str] | tuple[str, ...]) -> str:
     help=(
         "The path to a file containing the JSON Schema to use or an "
         "HTTP(S) URI for the schema. If a remote file is used, "
-        "it will be downloaded and cached locally based on mtime."
+        "it will be downloaded and cached locally based on mtime. "
+        "Use '-' for stdin."
     ),
+    metavar="[PATH|URI]",
 )
 @click.option(
     "--base-uri",

src/check_jsonschema/instance_loader.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import io
 import typing as t
 
 from .parsers import ParseError, ParserSet
@@ -25,14 +26,21 @@ def __init__(
 
     def iter_files(self) -> t.Iterator[tuple[str, ParseError | t.Any]]:
         for file in self._files:
-            if not hasattr(file, "name"):
+            if hasattr(file, "name"):
+                name = file.name
+            # allowing for BytesIO to be special-cased here is useful for
+            # simpler test setup, since this is what tests will pass and we naturally
+            # support it here
+            elif isinstance(file, io.BytesIO) or file.fileno() == 0:
+                name = "<stdin>"
+            else:
                 raise ValueError(f"File {file} has no name attribute")
             try:
                 data: t.Any = self._parsers.parse_data_with_path(
-                    file, file.name, self._default_filetype
+                    file, name, self._default_filetype
                 )
             except ParseError as err:
                 data = err
             else:
                 data = self._data_transform(data)
-            yield (file.name, data)
+            yield (name, data)

src/check_jsonschema/schema_loader/main.py

@@ -12,7 +12,7 @@
 from ..parsers import ParserSet
 from ..utils import is_url_ish
 from .errors import UnsupportedUrlScheme
-from .readers import HttpSchemaReader, LocalSchemaReader
+from .readers import HttpSchemaReader, LocalSchemaReader, StdinSchemaReader
 from .resolver import make_reference_registry
 
 
@@ -82,15 +82,22 @@ def __init__(
         self._parsers = ParserSet()
 
         # setup a schema reader lazily, when needed
-        self._reader: LocalSchemaReader | HttpSchemaReader | None = None
+        self._reader: LocalSchemaReader | HttpSchemaReader | StdinSchemaReader | None = (
+            None
+        )
 
     @property
-    def reader(self) -> LocalSchemaReader | HttpSchemaReader:
+    def reader(self) -> LocalSchemaReader | HttpSchemaReader | StdinSchemaReader:
         if self._reader is None:
             self._reader = self._get_schema_reader()
         return self._reader
 
-    def _get_schema_reader(self) -> LocalSchemaReader | HttpSchemaReader:
+    def _get_schema_reader(
+        self,
+    ) -> LocalSchemaReader | HttpSchemaReader | StdinSchemaReader:
+        if self.schemafile == "-":
+            return StdinSchemaReader()
+
         if self.url_info is None or self.url_info.scheme in ("file", ""):
             return LocalSchemaReader(self.schemafile)
 

src/check_jsonschema/schema_loader/readers.py

@@ -1,12 +1,14 @@
 from __future__ import annotations
 
 import io
+import json
+import sys
 import typing as t
 
 import ruamel.yaml
 
 from ..cachedownloader import CacheDownloader
-from ..parsers import ParserSet
+from ..parsers import ParseError, ParserSet
 from ..utils import filename2path
 from .errors import SchemaParseError
 
@@ -30,7 +32,7 @@ def __init__(self, filename: str) -> None:
         self.filename = str(self.path)
         self.parsers = ParserSet()
 
-    def get_retrieval_uri(self) -> str:
+    def get_retrieval_uri(self) -> str | None:
         return self.path.as_uri()
 
     def _read_impl(self) -> t.Any:
@@ -40,6 +42,20 @@ def read_schema(self) -> dict:
         return _run_load_callback(self.filename, self._read_impl)
 
 
+class StdinSchemaReader:
+    def __init__(self) -> None:
+        self.parsers = ParserSet()
+
+    def get_retrieval_uri(self) -> str | None:
+        return None
+
+    def read_schema(self) -> dict:
+        try:
+            return json.load(sys.stdin)
+        except ValueError as e:
+            raise ParseError("Failed to parse JSON from stdin") from e
+
+
 class HttpSchemaReader:
     def __init__(
         self,
@@ -64,7 +80,7 @@ def _parse(self, schema_bytes: bytes) -> t.Any:
             )
         return self._parsed_schema
 
-    def get_retrieval_uri(self) -> str:
+    def get_retrieval_uri(self) -> str | None:
         return self.url
 
     def _read_impl(self) -> t.Any:

tests/acceptance/test_special_filetypes.py

@@ -114,3 +114,37 @@ def _fake_compute_default_cache_dir(self):
         assert result.exit_code == 0
     else:
         assert result.exit_code == 1
+
+
+@pytest.mark.parametrize("check_passes", (True, False))
+@pytest.mark.parametrize("using_stdin", ("schema", "instance"))
+def test_schema_or_instance_from_stdin(
+    run_line, check_passes, tmp_path, monkeypatch, using_stdin
+):
+    """
+    a "remote schema" (meaning HTTPS) with bad data, therefore requiring that a retry
+    fires in order to parse
+    """
+    if using_stdin == "schema":
+        instance_path = tmp_path / "instance.json"
+        instance_path.write_text("42" if check_passes else '"foo"')
+
+        result = run_line(
+            ["check-jsonschema", "--schemafile", "-", str(instance_path)],
+            input='{"type": "integer"}',
+        )
+    elif using_stdin == "instance":
+        schema_path = tmp_path / "schema.json"
+        schema_path.write_text('{"type": "integer"}')
+        instance = "42" if check_passes else '"foo"'
+
+        result = run_line(
+            ["check-jsonschema", "--schemafile", schema_path, "-"],
+            input=instance,
+        )
+    else:
+        raise NotImplementedError
+    if check_passes:
+        assert result.exit_code == 0
+    else:
+        assert result.exit_code == 1