5226968556240896

Build Type

push

cirrus-ci

Committed by GitHub

Commit Message

Typo: validate-project => validate-pyproject (#78)

Run Details

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

95.43

/src/validate_pyproject/api.py

"""
Retrieve JSON schemas for validating dicts representing a ``pyproject.toml`` file.
"""
import json
import logging
import sys
from enum import Enum
from functools import partial, reduce
from itertools import chain
from types import MappingProxyType, ModuleType
from typing import (
    TYPE_CHECKING,
    Callable,
    Dict,
    Iterator,
    Mapping,
    Optional,
    Sequence,
    Tuple,
    TypeVar,
    Union,
    cast,
)

from . import errors, formats
from ._vendor import fastjsonschema as FJS
from .error_reporting import detailed_errors
from .extra_validations import EXTRA_VALIDATIONS
from .types import FormatValidationFn, Schema, ValidationFn

_logger = logging.getLogger(__name__)
_chain_iter = chain.from_iterable

if TYPE_CHECKING:  # pragma: no cover
    from .plugins import PluginWrapper  # noqa


try:  # pragma: no cover
    if sys.version_info[:2] < (3, 7) or TYPE_CHECKING:  # See #22
        from importlib_resources import files
    else:
        from importlib.resources import files

    def read_text(package: Union[str, ModuleType], resource) -> str:
        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")
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.
    """
    return Schema(json.loads(read_text(package, f"{name}{ext}")))


def load_builtin_plugin(name: str) -> Schema:
    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``.
    """

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

        top_level = cast(dict, load(TOP_LEVEL_SCHEMA))  # Make it mutable
        self._spec_version = 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:
            pid, tool, schema = plugin.id, plugin.tool, plugin.schema
            if plugin.tool in tool_properties:
                _logger.warning(f"{plugin.id} overwrites `tool.{plugin.tool}` schema")
            else:
                _logger.info(f"{pid} defines `tool.{tool}` schema")
            sid = self._ensure_compatibility(tool, schema)["$id"]
            tool_properties[tool] = {"$ref": sid}
            self._schemas[sid] = (f"tool.{tool}", pid, schema)

        self._main_id = sid = top_level["$id"]
        main_schema = Schema(top_level)
        origin = f"{__name__} - build metadata"
        self._schemas[sid] = ("<$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) -> Schema:
        if "$id" not in schema:
            raise errors.SchemaMissingId(reference)
        sid = schema["$id"]
        if sid in self._schemas:
            raise errors.SchemaWithDuplicatedId(sid)
        version = schema.get("$schema")
        if version and version != self.spec_version:
            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.
    """

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

    def __contains__(self, key) -> bool:
        return_val = isinstance(key, str)
        if return_val and key not in self._uri_schemas:
            self._uri_schemas.append(key)
        return return_val

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

    def __len__(self):
        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:
    def __init__(
        self,
        plugins: Union[Sequence["PluginWrapper"], AllPlugins] = ALL_PLUGINS,
        format_validators: Mapping[str, FormatValidationFn] = FORMAT_FUNCTIONS,
        extra_validations: Sequence[ValidationFn] = EXTRA_VALIDATIONS,
    ):
        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

            self._plugins = tuple(list_from_entry_points())
        else:
            self._plugins = tuple(plugins)  # force immutability / read only

        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)

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

abravalheri / validate-pyproject / 5226968556240896

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