Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -80,6 +80,7 @@ profile = "black"

[tool.mypy]
# strict = true # TODO: enable
disallow_any_generics = true
disallow_untyped_defs = true
ignore_missing_imports = true
warn_unreachable = true
Expand Down
10 changes: 6 additions & 4 deletions src/check_jsonschema/cli/param_types.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@ def wrapper(*args: t.Any, **kwargs: t.Any) -> None:
return wrapper # type: ignore[return-value]


class CommaDelimitedList(click.ParamType):
class CommaDelimitedList(click.ParamType[list[str]]):
name = "comma_delimited"

def __init__(
Expand All @@ -46,7 +46,8 @@ def get_metavar(self, param: click.Parameter, ctx: click.Context | None) -> str:
def convert(
self, value: str, param: click.Parameter | None, ctx: click.Context | None
) -> list[str]:
value = super().convert(value, param, ctx)
# ignore assignment error (super().convert() returns `str`, not `list[str]`)
value = super().convert(value, param, ctx) # type: ignore[assignment]

# if `--foo` is a comma delimited list and someone passes
# `--foo ""`, take that as `foo=[]` rather than foo=[""]
Expand All @@ -67,7 +68,7 @@ def convert(
return resolved


class ValidatorClassName(click.ParamType):
class ValidatorClassName(click.ParamType[type[jsonschema.protocols.Validator]]):
name = "validator"

def convert(
Expand All @@ -89,7 +90,8 @@ def convert(
As a result, only one import is needed, rather than iterative imports over the
list of names.
"""
value = super().convert(value, param, ctx)
# ignore type error: super().convert() returns str, not a validator class
value = super().convert(value, param, ctx) # type: ignore[assignment]
pattern = re.compile(
r"^(?P<pkg>(?!\d)(\w+)(\.(?!\d)(\w+))*):"
r"(?P<cls>(?!\d)(\w+)(\.(?!\d)(\w+))*)$"
Expand Down
4 changes: 2 additions & 2 deletions src/check_jsonschema/parsers/json5.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,7 @@
import pyjson5

ParseError: type[Exception] = pyjson5.Json5DecoderException
_load: t.Callable | None = pyjson5.load
_load: t.Callable[[t.IO[bytes]], t.Any] | None = pyjson5.load
except ImportError:
# if pyjson5 was not available, try to import 'json5', the pure-python implementation
try:
Expand All @@ -25,7 +25,7 @@
ENABLED = _load is not None

if _load is not None:
_load_concrete: t.Callable = _load
_load_concrete: t.Callable[[t.IO[bytes]], t.Any] = _load

def load(stream: t.IO[bytes]) -> t.Any:
return _load_concrete(stream)
Expand Down
4 changes: 2 additions & 2 deletions src/check_jsonschema/reporter.py
Original file line number Diff line number Diff line change
Expand Up @@ -157,7 +157,7 @@ def report_success(self, result: CheckResult) -> None:
def _dump_error_map(
self,
error_map: dict[str, list[jsonschema.ValidationError]],
) -> t.Iterator[dict]:
) -> t.Iterator[dict[str, t.Any]]:
for filename, errors in error_map.items():
for err in errors:
item = {
Expand Down Expand Up @@ -190,7 +190,7 @@ def _dump_error_map(
def _dump_parse_errors(
self,
error_map: dict[str, list[ParseError]],
) -> t.Iterator[dict]:
) -> t.Iterator[dict[str, str]]:
for filename, errors in error_map.items():
for err in errors:
yield {
Expand Down
34 changes: 20 additions & 14 deletions src/check_jsonschema/schema_loader/readers.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
from __future__ import annotations

import enum
import io
import json
import sys
Expand All @@ -15,14 +16,16 @@
yaml = ruamel.yaml.YAML(typ="safe")


class _UnsetType:
pass
class _UnsetType(enum.Enum):
UNSET = enum.auto


_UNSET = _UnsetType()
_UNSET = _UnsetType.UNSET


def _run_load_callback(schema_location: str, callback: t.Callable) -> dict:
def _run_load_callback(
schema_location: str, callback: t.Callable[[], dict[str, t.Any]]
) -> dict[str, t.Any]:
try:
schema = callback()
# only local loads can raise the YAMLError, but catch for both cases for simplicity
Expand All @@ -38,35 +41,38 @@ def __init__(self, filename: str) -> None:
self.path = filename2path(filename)
self.filename = str(self.path)
self.parsers = ParserSet()
self._parsed_schema: dict | _UnsetType = _UNSET
self._parsed_schema: dict[str, t.Any] | t.Literal[_UnsetType.UNSET] = _UNSET

def get_retrieval_uri(self) -> str | None:
return self.path.as_uri()

def _read_impl(self) -> t.Any:
return self.parsers.parse_file(self.path, default_filetype="json")

def read_schema(self) -> dict:
def read_schema(self) -> dict[str, t.Any]:
if self._parsed_schema is _UNSET:
self._parsed_schema = _run_load_callback(self.filename, self._read_impl)
return t.cast(dict, self._parsed_schema)
return self._parsed_schema


class StdinSchemaReader:
def __init__(self) -> None:
self.parsers = ParserSet()
self._parsed_schema: dict | _UnsetType = _UNSET
self._parsed_schema: dict[str, t.Any] | t.Literal[_UnsetType.UNSET] = _UNSET

def get_retrieval_uri(self) -> str | None:
return None

def read_schema(self) -> dict:
def read_schema(self) -> dict[str, t.Any]:
if self._parsed_schema is _UNSET:
try:
self._parsed_schema = json.load(sys.stdin)
loaded: dict[str, t.Any] = json.load(sys.stdin)
except ValueError as e:
raise ParseError("Failed to parse JSON from stdin") from e
return t.cast(dict, self._parsed_schema)
self._parsed_schema = loaded
return self._parsed_schema
else:
return self._parsed_schema


class HttpSchemaReader:
Expand All @@ -80,7 +86,7 @@ def __init__(
self.downloader = CacheDownloader("schemas", disable_cache=disable_cache).bind(
url, validation_callback=self._parse
)
self._parsed_schema: dict | _UnsetType = _UNSET
self._parsed_schema: dict[str, t.Any] | t.Literal[_UnsetType.UNSET] = _UNSET

def _parse(self, schema_bytes: bytes) -> t.Any:
return self.parsers.parse_data_with_path(
Expand All @@ -94,7 +100,7 @@ def _read_impl(self) -> t.Any:
with self.downloader.open() as fp:
return self._parse(fp.read())

def read_schema(self) -> dict:
def read_schema(self) -> dict[str, t.Any]:
if self._parsed_schema is _UNSET:
self._parsed_schema = _run_load_callback(self.url, self._read_impl)
return t.cast(dict, self._parsed_schema)
return self._parsed_schema
5 changes: 4 additions & 1 deletion src/check_jsonschema/schema_loader/resolver.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,10 @@


def make_reference_registry(
parsers: ParserSet, retrieval_uri: str | None, schema: dict, disable_cache: bool
parsers: ParserSet,
retrieval_uri: str | None,
schema: dict[str, t.Any],
disable_cache: bool,
) -> referencing.Registry:
id_attribute_: t.Any = schema.get("$id")
if isinstance(id_attribute_, str):
Expand Down
8 changes: 5 additions & 3 deletions src/check_jsonschema/transforms/azure_pipelines.py
Original file line number Diff line number Diff line change
Expand Up @@ -52,7 +52,7 @@ def traverse_data(data: t.Any) -> t.Any:
return data


def traverse_list(data: list) -> list:
def traverse_list(data: list[t.Any]) -> list[t.Any]:
ret = []
for item in data:
# is the current item a single-value dict with an expression as its key?
Expand All @@ -77,7 +77,7 @@ def traverse_list(data: list) -> list:
return ret


def traverse_dict(data: dict) -> dict:
def traverse_dict(data: dict[t.Any, t.Any]) -> dict[t.Any, t.Any]:
newdata = {}
for key, value in data.items():
newvalue = traverse_data(value)
Expand Down Expand Up @@ -122,7 +122,9 @@ def traverse_dict(data: dict) -> dict:
return newdata


def azure_main(data: dict | list) -> dict | list:
def azure_main(
data: dict[t.Any, t.Any] | list[t.Any],
) -> dict[t.Any, t.Any] | list[t.Any]:
if isinstance(data, list):
raise AzurePipelinesDataError(
"this transform requires that the data be an object, got list"
Expand Down
6 changes: 4 additions & 2 deletions src/check_jsonschema/transforms/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,19 +4,21 @@

import ruamel.yaml

_AnyData: t.TypeAlias = list[t.Any] | dict[t.Any, t.Any]


class Transform:
def __init__(
self,
*,
on_data: t.Callable[[list | dict], list | dict] | None = None,
on_data: t.Callable[[_AnyData], _AnyData] | None = None,
) -> None:
self.on_data = on_data

def modify_yaml_implementation(self, implementation: ruamel.yaml.YAML) -> None:
pass

def __call__(self, data: list | dict) -> list | dict:
def __call__(self, data: _AnyData) -> _AnyData:
if self.on_data is not None:
return self.on_data(data)
return data
Loading