Support reading schemafiles and instancefiles from stdin

src/check_jsonschema/checker.py

@@ -47,7 +47,7 @@ def _fail(self, msg: str, err: Exception | None = None) -> t.NoReturn:
         raise _Exit(1)
 
     def get_validator(
-        self, path: pathlib.Path, doc: dict[str, t.Any]
+        self, path: pathlib.Path | str, doc: dict[str, t.Any]
     ) -> jsonschema.protocols.Validator:
         try:
             return self._schema_loader.get_validator(

src/check_jsonschema/cli/main_command.py

@@ -2,6 +2,7 @@
 
 import os
 import textwrap
+import typing as t
 
 import click
 import jsonschema
@@ -90,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",
@@ -217,7 +220,7 @@ def pretty_helptext_list(values: list[str] | tuple[str, ...]) -> str:
     help="Reduce output verbosity",
     count=True,
 )
-@click.argument("instancefiles", required=True, nargs=-1)
+@click.argument("instancefiles", required=True, nargs=-1, type=click.File("rb"))
 def main(
     *,
     schemafile: str | None,
@@ -236,7 +239,7 @@ def main(
     output_format: str,
     verbose: int,
     quiet: int,
-    instancefiles: tuple[str, ...],
+    instancefiles: tuple[t.BinaryIO, ...],
 ) -> None:
     args = ParseResult()
 

src/check_jsonschema/cli/parse_result.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import enum
+import typing as t
 
 import click
 import jsonschema
@@ -21,7 +22,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[str, ...] = ()
+        self.instancefiles: tuple[t.BinaryIO, ...] = ()
         # cache controls
         self.disable_cache: bool = False
         self.cache_filename: str | None = None

src/check_jsonschema/instance_loader.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-import pathlib
+import io
 import typing as t
 
 from .parsers import ParseError, ParserSet
@@ -10,11 +10,11 @@
 class InstanceLoader:
     def __init__(
         self,
-        filenames: t.Sequence[str],
+        files: t.Sequence[t.BinaryIO],
         default_filetype: str = "json",
         data_transform: Transform | None = None,
     ) -> None:
-        self._filenames = filenames
+        self._files = files
         self._default_filetype = default_filetype
         self._data_transform = (
             data_transform if data_transform is not None else Transform()
@@ -24,13 +24,23 @@ def __init__(
             modify_yaml_implementation=self._data_transform.modify_yaml_implementation
         )
 
-    def iter_files(self) -> t.Iterator[tuple[pathlib.Path, ParseError | t.Any]]:
-        for fn in self._filenames:
-            path = pathlib.Path(fn)
+    def iter_files(self) -> t.Iterator[tuple[str, ParseError | t.Any]]:
+        for file in self._files:
+            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_file(path, self._default_filetype)
+                data: t.Any = self._parsers.parse_data_with_path(
+                    file, name, self._default_filetype
+                )
             except ParseError as err:
                 data = err
             else:
                 data = self._data_transform(data)
-            yield (path, data)
+            yield (name, data)

src/check_jsonschema/result.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import pathlib
 
 import jsonschema
@@ -15,18 +17,18 @@ def __init__(self) -> None:
     def success(self) -> bool:
         return not (bool(self.parse_errors) or bool(self.validation_errors))
 
-    def record_validation_success(self, path: pathlib.Path) -> None:
+    def record_validation_success(self, path: pathlib.Path | str) -> None:
         self.successes.append(str(path))
 
     def record_validation_error(
-        self, path: pathlib.Path, err: jsonschema.ValidationError
+        self, path: pathlib.Path | str, err: jsonschema.ValidationError
     ) -> None:
         filename = str(path)
         if filename not in self.validation_errors:
             self.validation_errors[filename] = []
         self.validation_errors[filename].append(err)
 
-    def record_parse_error(self, path: pathlib.Path, err: ParseError) -> None:
+    def record_parse_error(self, path: pathlib.Path | str, err: ParseError) -> None:
         filename = str(path)
         if filename not in self.parse_errors:
             self.parse_errors[filename] = []

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
 
 
@@ -47,7 +47,7 @@ def set_defaults_then_validate(
 class SchemaLoaderBase:
     def get_validator(
         self,
-        path: pathlib.Path,
+        path: pathlib.Path | str,
         instance_doc: dict[str, t.Any],
         format_opts: FormatOptions,
         fill_defaults: bool,
@@ -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)
 
@@ -117,7 +124,7 @@ def get_schema(self) -> dict[str, t.Any]:
 
     def get_validator(
         self,
-        path: pathlib.Path,
+        path: pathlib.Path | str,
         instance_doc: dict[str, t.Any],
         format_opts: FormatOptions,
         fill_defaults: bool,
@@ -189,7 +196,7 @@ def __init__(self, base_uri: str | None = None) -> None:
 
     def get_validator(
         self,
-        path: pathlib.Path,
+        path: pathlib.Path | str,
         instance_doc: dict[str, t.Any],
         format_opts: FormatOptions,
         fill_defaults: bool,

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

tests/conftest.py

@@ -1,3 +1,4 @@
+import os
 import pathlib
 import sys
 
@@ -38,3 +39,10 @@ def func(path, text):
     for name in all_names_to_clear:
         if name in sys.modules:
             del sys.modules[name]
+
+
+@pytest.fixture
+def in_tmp_dir(request, tmp_path):
+    os.chdir(str(tmp_path))
+    yield
+    os.chdir(request.config.invocation_dir)