6173991897923584

Committed 11 Nov 2024 04:41PM CUT coverage: 97.859%. Remained the same

Build # 6173991897923584

Build Type

Pull #218

cirrus-ci

Committed by

pre-commit-ci[bot]

Commit Message

[pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

Pull Request Pull Request #218: [pre-commit.ci] pre-commit autoupdate

Run Details

293 of 306 branches covered (95.75%)

Branch coverage included in aggregate %.

1 of 1 new or added line in 1 file covered. (100.0%)

941 of 955 relevant lines covered (98.53%)

6.89 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

94.25

/src/validate_pyproject/api.py

"""
Retrieve JSON schemas for validating dicts representing a ``pyproject.toml`` file.
"""

import json
import logging
import typing
from enum import Enum
from functools import partial, reduce
from types import MappingProxyType, ModuleType
from typing import (
    Callable,
    Dict,
    Iterator,
    Mapping,
    Optional,
    Sequence,
    Tuple,
    TypeVar,
    Union,
)

import fastjsonschema as FJS

from . import errors, formats
from .error_reporting import detailed_errors
from .extra_validations import EXTRA_VALIDATIONS
from .types import FormatValidationFn, Schema, ValidationFn

_logger = logging.getLogger(__name__)

if typing.TYPE_CHECKING:  # pragma: no cover
    from .plugins import PluginProtocol


try:  # pragma: no cover
    from importlib.resources import files

    def read_text(package: Union[str, ModuleType], resource: str) -> str:
        """:meta private:"""
        return files(package).joinpath(resource).read_text(encoding="utf-8")

except ImportError:  # pragma: no cover
    from importlib.resources import read_text


T = TypeVar("T", bound=Mapping)
AllPlugins = Enum("AllPlugins", "ALL_PLUGINS")  #: :meta private:
ALL_PLUGINS = AllPlugins.ALL_PLUGINS

TOP_LEVEL_SCHEMA = "pyproject_toml"
PROJECT_TABLE_SCHEMA = "project_metadata"


def _get_public_functions(module: ModuleType) -> Mapping[str, FormatValidationFn]:
    return {
        fn.__name__.replace("_", "-"): fn
        for fn in module.__dict__.values()
        if callable(fn) and not fn.__name__.startswith("_")
    }


FORMAT_FUNCTIONS = MappingProxyType(_get_public_functions(formats))


def load(name: str, package: str = __package__, ext: str = ".schema.json") -> Schema:
    """Load the schema from a JSON Schema file.
    The returned dict-like object is immutable.

    :meta private: (low level detail)
    """
    return Schema(json.loads(read_text(package, f"{name}{ext}")))


def load_builtin_plugin(name: str) -> Schema:
    """:meta private: (low level detail)"""
    return load(name, f"{__package__}.plugins")


class SchemaRegistry(Mapping[str, Schema]):
    """Repository of parsed JSON Schemas used for validating a ``pyproject.toml``.

    During instantiation the schemas equivalent to PEP 517, PEP 518 and PEP 621
    will be combined with the schemas for the ``tool`` subtables provided by the
    plugins.

    Since this object work as a mapping between each schema ``$id`` and the schema
    itself, all schemas provided by plugins **MUST** have a top level ``$id``.

    :meta private: (low level detail)
    """

    def __init__(self, plugins: Sequence["PluginProtocol"] = ()):
        self._schemas: Dict[str, Tuple[str, str, Schema]] = {}
        # (which part of the TOML, who defines, schema)

        top_level = typing.cast(dict, load(TOP_LEVEL_SCHEMA))  # Make it mutable
        self._spec_version: str = top_level["$schema"]
        top_properties = top_level["properties"]
        tool_properties = top_properties["tool"].setdefault("properties", {})

        # Add PEP 621
        project_table_schema = load(PROJECT_TABLE_SCHEMA)
        self._ensure_compatibility(PROJECT_TABLE_SCHEMA, project_table_schema)
        sid = project_table_schema["$id"]
        top_level["project"] = {"$ref": sid}
        origin = f"{__name__} - project metadata"
        self._schemas = {sid: ("project", origin, project_table_schema)}

        # Add tools using Plugins
        for plugin in plugins:
            allow_overwrite: Optional[str] = None
            if plugin.tool in tool_properties:
                _logger.warning(f"{plugin.id} overwrites `tool.{plugin.tool}` schema")
                allow_overwrite = plugin.schema.get("$id")
            else:
                _logger.info(f"{plugin.id} defines `tool.{plugin.tool}` schema")
            compatible = self._ensure_compatibility(
                plugin.tool, plugin.schema, allow_overwrite
            )
            sid = compatible["$id"]
            sref = f"{sid}#{plugin.fragment}" if plugin.fragment else sid
            tool_properties[plugin.tool] = {"$ref": sref}
            self._schemas[sid] = (f"tool.{plugin.tool}", plugin.id, plugin.schema)

        self._main_id: str = top_level["$id"]
        main_schema = Schema(top_level)
        origin = f"{__name__} - build metadata"
        self._schemas[self._main_id] = ("<$ROOT>", origin, main_schema)

    @property
    def spec_version(self) -> str:
        """Version of the JSON Schema spec in use"""
        return self._spec_version

    @property
    def main(self) -> str:
        """Top level schema for validating a ``pyproject.toml`` file"""
        return self._main_id

    def _ensure_compatibility(
        self, reference: str, schema: Schema, allow_overwrite: Optional[str] = None
    ) -> Schema:
        if "$id" not in schema or not schema["$id"]:
            raise errors.SchemaMissingId(reference)
        sid = schema["$id"]
        if sid in self._schemas and sid != allow_overwrite:
            raise errors.SchemaWithDuplicatedId(sid)
        version = schema.get("$schema")
        # Support schemas with missing trailing # (incorrect, but required before 0.15)
        if version and version.rstrip("#") != self.spec_version.rstrip("#"):
            raise errors.InvalidSchemaVersion(reference, version, self.spec_version)
        return schema

    def __getitem__(self, key: str) -> Schema:
        return self._schemas[key][-1]

    def __iter__(self) -> Iterator[str]:
        return iter(self._schemas)

    def __len__(self) -> int:
        return len(self._schemas)


class RefHandler(Mapping[str, Callable[[str], Schema]]):
    """:mod:`fastjsonschema` allows passing a dict-like object to load external schema
    ``$ref``s. Such objects map the URI schema (e.g. ``http``, ``https``, ``ftp``)
    into a function that receives the schema URI and returns the schema (as parsed JSON)
    (otherwise :mod:`urllib` is used and the URI is assumed to be a valid URL).
    This class will ensure all the URIs are loaded from the local registry.

    :meta private: (low level detail)
    """

    def __init__(self, registry: Mapping[str, Schema]):
        self._uri_schemas = ["http", "https"]
        self._registry = registry

    def __contains__(self, key: object) -> bool:
        if isinstance(key, str):
            if key not in self._uri_schemas:
                self._uri_schemas.append(key)
            return True
        return False

    def __iter__(self) -> Iterator[str]:
        return iter(self._uri_schemas)

    def __len__(self) -> int:
        return len(self._uri_schemas)

    def __getitem__(self, key: str) -> Callable[[str], Schema]:
        """All the references should be retrieved from the registry"""
        return self._registry.__getitem__


class Validator:
    _plugins: Sequence["PluginProtocol"]

    def __init__(
        self,
        plugins: Union[Sequence["PluginProtocol"], AllPlugins] = ALL_PLUGINS,
        format_validators: Mapping[str, FormatValidationFn] = FORMAT_FUNCTIONS,
        extra_validations: Sequence[ValidationFn] = EXTRA_VALIDATIONS,
        *,
        extra_plugins: Sequence["PluginProtocol"] = (),
    ):
        self._code_cache: Optional[str] = None
        self._cache: Optional[ValidationFn] = None
        self._schema: Optional[Schema] = None

        # Let's make the following options readonly
        self._format_validators = MappingProxyType(format_validators)
        self._extra_validations = tuple(extra_validations)

        if plugins is ALL_PLUGINS:
            from .plugins import list_from_entry_points

            plugins = list_from_entry_points()

        self._plugins = (*plugins, *extra_plugins)

        self._schema_registry = SchemaRegistry(self._plugins)
        self.handlers = RefHandler(self._schema_registry)

    @property
    def registry(self) -> SchemaRegistry:
        return self._schema_registry

    @property
    def schema(self) -> Schema:
        """Top level ``pyproject.toml`` JSON Schema"""
        return Schema({"$ref": self._schema_registry.main})

    @property
    def extra_validations(self) -> Sequence[ValidationFn]:
        """List of extra validation functions that run after the JSON Schema check"""
        return self._extra_validations

    @property
    def formats(self) -> Mapping[str, FormatValidationFn]:
        """Mapping between JSON Schema formats and functions that validates them"""
        return self._format_validators

    @property
    def generated_code(self) -> str:
        if self._code_cache is None:
            fmts = dict(self.formats)
            self._code_cache = FJS.compile_to_code(
                self.schema, self.handlers, fmts, use_default=False
            )

        return self._code_cache

    def __getitem__(self, schema_id: str) -> Schema:
        """Retrieve a schema from registry"""
        return self._schema_registry[schema_id]

    def __call__(self, pyproject: T) -> T:
        """Checks a parsed ``pyproject.toml`` file (given as :obj:`typing.Mapping`)
        and raises an exception when it is not a valid.
        """
        if self._cache is None:
            compiled = FJS.compile(
                self.schema, self.handlers, dict(self.formats), use_default=False
            )
            fn = partial(compiled, custom_formats=self._format_validators)
            self._cache = typing.cast(ValidationFn, fn)

        with detailed_errors():
            self._cache(pyproject)
            return reduce(lambda acc, fn: fn(acc), self.extra_validations, pyproject)

1	"""
2	Retrieve JSON schemas for validating dicts representing a ``pyproject.toml`` file.
3	"""
4
5	import json	7✔
6	import logging	7✔
7	import typing	7✔
8	from enum import Enum	7✔
9	from functools import partial, reduce	7✔
10	from types import MappingProxyType, ModuleType	7✔
11	from typing import (	7✔
12	Callable,
13	Dict,
14	Iterator,
15	Mapping,
16	Optional,
17	Sequence,
18	Tuple,
19	TypeVar,
20	Union,
21	)
22
23	import fastjsonschema as FJS	7✔
24
25	from . import errors, formats	7✔
26	from .error_reporting import detailed_errors	7✔
27	from .extra_validations import EXTRA_VALIDATIONS	7✔
28	from .types import FormatValidationFn, Schema, ValidationFn	7✔
29
30	_logger = logging.getLogger(__name__)	7✔
31
32	if typing.TYPE_CHECKING: # pragma: no cover
33	from .plugins import PluginProtocol
34
35
36	try: # pragma: no cover
37	from importlib.resources import files
38
39	def read_text(package: Union[str, ModuleType], resource: str) -> str:
40	""":meta private:"""
41	return files(package).joinpath(resource).read_text(encoding="utf-8")
42
43	except ImportError: # pragma: no cover
44	from importlib.resources import read_text
45
46
47	T = TypeVar("T", bound=Mapping)	7✔
48	AllPlugins = Enum("AllPlugins", "ALL_PLUGINS") #: :meta private:	7✔
49	ALL_PLUGINS = AllPlugins.ALL_PLUGINS	7✔
50
51	TOP_LEVEL_SCHEMA = "pyproject_toml"	7✔
52	PROJECT_TABLE_SCHEMA = "project_metadata"	7✔
53
54
55	def _get_public_functions(module: ModuleType) -> Mapping[str, FormatValidationFn]:	7✔
56	return {	7✔
57	fn.__name__.replace("_", "-"): fn
58	for fn in module.__dict__.values()
59	if callable(fn) and not fn.__name__.startswith("_")
60	}
61
62
63	FORMAT_FUNCTIONS = MappingProxyType(_get_public_functions(formats))	7✔
64
65
66	def load(name: str, package: str = __package__, ext: str = ".schema.json") -> Schema:	7✔
67	"""Load the schema from a JSON Schema file.
68	The returned dict-like object is immutable.
69
70	:meta private: (low level detail)
71	"""
72	return Schema(json.loads(read_text(package, f"{name}{ext}")))	7✔
73
74
75	def load_builtin_plugin(name: str) -> Schema:	7✔
76	""":meta private: (low level detail)"""
77	return load(name, f"{__package__}.plugins")	7✔
78
79
80	class SchemaRegistry(Mapping[str, Schema]):	7✔
81	"""Repository of parsed JSON Schemas used for validating a ``pyproject.toml``.
82
83	During instantiation the schemas equivalent to PEP 517, PEP 518 and PEP 621
84	will be combined with the schemas for the ``tool`` subtables provided by the
85	plugins.
86
87	Since this object work as a mapping between each schema ``$id`` and the schema
88	itself, all schemas provided by plugins MUST have a top level ``$id``.
89
90	:meta private: (low level detail)
91	"""
92
93	def __init__(self, plugins: Sequence["PluginProtocol"] = ()):	7✔
94	self._schemas: Dict[str, Tuple[str, str, Schema]] = {}	7✔
95	# (which part of the TOML, who defines, schema)
96
97	top_level = typing.cast(dict, load(TOP_LEVEL_SCHEMA)) # Make it mutable	7✔
98	self._spec_version: str = top_level["$schema"]	7✔
99	top_properties = top_level["properties"]	7✔
100	tool_properties = top_properties["tool"].setdefault("properties", {})	7✔
101
102	# Add PEP 621
103	project_table_schema = load(PROJECT_TABLE_SCHEMA)	7✔
104	self._ensure_compatibility(PROJECT_TABLE_SCHEMA, project_table_schema)	7✔
105	sid = project_table_schema["$id"]	7✔
106	top_level["project"] = {"$ref": sid}	7✔
107	origin = f"{__name__} - project metadata"	7✔
108	self._schemas = {sid: ("project", origin, project_table_schema)}	7✔
109
110	# Add tools using Plugins
111	for plugin in plugins:	7✔
112	allow_overwrite: Optional[str] = None	7✔
113	if plugin.tool in tool_properties:	7✔
114	_logger.warning(f"{plugin.id} overwrites `tool.{plugin.tool}` schema")	7✔
115	allow_overwrite = plugin.schema.get("$id")	7✔
116	else:
117	_logger.info(f"{plugin.id} defines `tool.{plugin.tool}` schema")	7✔
118	compatible = self._ensure_compatibility(	7✔
119	plugin.tool, plugin.schema, allow_overwrite
120	)
121	sid = compatible["$id"]	7✔
122	sref = f"{sid}#{plugin.fragment}" if plugin.fragment else sid	7✔
123	tool_properties[plugin.tool] = {"$ref": sref}	7✔
124	self._schemas[sid] = (f"tool.{plugin.tool}", plugin.id, plugin.schema)	7✔
125
126	self._main_id: str = top_level["$id"]	7✔
127	main_schema = Schema(top_level)	7✔
128	origin = f"{__name__} - build metadata"	7✔
129	self._schemas[self._main_id] = ("<$ROOT>", origin, main_schema)	7✔
130
131	@property	7✔
132	def spec_version(self) -> str:	7✔
133	"""Version of the JSON Schema spec in use"""
134	return self._spec_version	7✔
135
136	@property	7✔
137	def main(self) -> str:	7✔
138	"""Top level schema for validating a ``pyproject.toml`` file"""
139	return self._main_id	7✔
140
141	def _ensure_compatibility(	7✔
142	self, reference: str, schema: Schema, allow_overwrite: Optional[str] = None
143	) -> Schema:
144	if "$id" not in schema or not schema["$id"]:	7✔
145	raise errors.SchemaMissingId(reference)	7✔
146	sid = schema["$id"]	7✔
147	if sid in self._schemas and sid != allow_overwrite:	7✔
148	raise errors.SchemaWithDuplicatedId(sid)	7✔
149	version = schema.get("$schema")	7✔
150	# Support schemas with missing trailing # (incorrect, but required before 0.15)
151	if version and version.rstrip("#") != self.spec_version.rstrip("#"):	7✔
152	raise errors.InvalidSchemaVersion(reference, version, self.spec_version)	7✔
153	return schema	7✔
154
155	def __getitem__(self, key: str) -> Schema:	7✔
156	return self._schemas[key][-1]	7✔
157
158	def __iter__(self) -> Iterator[str]:	7✔
159	return iter(self._schemas)	×
160
161	def __len__(self) -> int:	7✔
162	return len(self._schemas)	×
163
164
165	class RefHandler(Mapping[str, Callable[[str], Schema]]):	7✔
166	""":mod:`fastjsonschema` allows passing a dict-like object to load external schema
167	``$ref``s. Such objects map the URI schema (e.g. ``http``, ``https``, ``ftp``)
168	into a function that receives the schema URI and returns the schema (as parsed JSON)
169	(otherwise :mod:`urllib` is used and the URI is assumed to be a valid URL).
170	This class will ensure all the URIs are loaded from the local registry.
171
172	:meta private: (low level detail)
173	"""
174
175	def __init__(self, registry: Mapping[str, Schema]):	7✔
176	self._uri_schemas = ["http", "https"]	7✔
177	self._registry = registry	7✔
178
179	def __contains__(self, key: object) -> bool:	7✔
180	if isinstance(key, str):	7!
181	if key not in self._uri_schemas:	7!
182	self._uri_schemas.append(key)	×
183	return True	7✔
184	return False	×
185
186	def __iter__(self) -> Iterator[str]:	7✔
187	return iter(self._uri_schemas)	×
188
189	def __len__(self) -> int:	7✔
190	return len(self._uri_schemas)	×
191
192	def __getitem__(self, key: str) -> Callable[[str], Schema]:	7✔
193	"""All the references should be retrieved from the registry"""
194	return self._registry.__getitem__	7✔
195
196
197	class Validator:	7✔
198	_plugins: Sequence["PluginProtocol"]	7✔
199
200	def __init__(	7✔
201	self,
202	plugins: Union[Sequence["PluginProtocol"], AllPlugins] = ALL_PLUGINS,
203	format_validators: Mapping[str, FormatValidationFn] = FORMAT_FUNCTIONS,
204	extra_validations: Sequence[ValidationFn] = EXTRA_VALIDATIONS,
205	*,
206	extra_plugins: Sequence["PluginProtocol"] = (),
207	):
208	self._code_cache: Optional[str] = None	7✔
209	self._cache: Optional[ValidationFn] = None	7✔
210	self._schema: Optional[Schema] = None	7✔
211
212	# Let's make the following options readonly
213	self._format_validators = MappingProxyType(format_validators)	7✔
214	self._extra_validations = tuple(extra_validations)	7✔
215
216	if plugins is ALL_PLUGINS:	7✔
217	from .plugins import list_from_entry_points	7✔
218
219	plugins = list_from_entry_points()	7✔
220
221	self._plugins = (plugins, extra_plugins)	7✔
222
223	self._schema_registry = SchemaRegistry(self._plugins)	7✔
224	self.handlers = RefHandler(self._schema_registry)	7✔
225
226	@property	7✔
227	def registry(self) -> SchemaRegistry:	7✔
228	return self._schema_registry	7✔
229
230	@property	7✔
231	def schema(self) -> Schema:	7✔
232	"""Top level ``pyproject.toml`` JSON Schema"""
233	return Schema({"$ref": self._schema_registry.main})	7✔
234
235	@property	7✔
236	def extra_validations(self) -> Sequence[ValidationFn]:	7✔
237	"""List of extra validation functions that run after the JSON Schema check"""
238	return self._extra_validations	7✔
239
240	@property	7✔
241	def formats(self) -> Mapping[str, FormatValidationFn]:	7✔
242	"""Mapping between JSON Schema formats and functions that validates them"""
243	return self._format_validators	7✔
244
245	@property	7✔
246	def generated_code(self) -> str:	7✔
247	if self._code_cache is None:	7!
248	fmts = dict(self.formats)	7✔
249	self._code_cache = FJS.compile_to_code(	7✔
250	self.schema, self.handlers, fmts, use_default=False
251	)
252
253	return self._code_cache	7✔
254
255	def __getitem__(self, schema_id: str) -> Schema:	7✔
256	"""Retrieve a schema from registry"""
257	return self._schema_registry[schema_id]	×
258
259	def __call__(self, pyproject: T) -> T:	7✔
260	"""Checks a parsed ``pyproject.toml`` file (given as :obj:`typing.Mapping`)
261	and raises an exception when it is not a valid.
262	"""
263	if self._cache is None:	7✔
264	compiled = FJS.compile(	7✔
265	self.schema, self.handlers, dict(self.formats), use_default=False
266	)
267	fn = partial(compiled, custom_formats=self._format_validators)	7✔
268	self._cache = typing.cast(ValidationFn, fn)	7✔
269
270	with detailed_errors():	7✔
271	self._cache(pyproject)	7✔
272	return reduce(lambda acc, fn: fn(acc), self.extra_validations, pyproject)	7✔

abravalheri / validate-pyproject / 6173991897923584

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous