• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

abravalheri / validate-pyproject / 11034501584

25 Sep 2024 01:49PM CUT coverage: 97.823% (-0.2%) from 97.976%
11034501584

push

github

abravalheri
Prevent Github action for ignoring files for cache

551 of 571 branches covered (96.5%)

Branch coverage included in aggregate %.

932 of 945 relevant lines covered (98.62%)

5.91 hits per line

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

93.6
/src/validate_pyproject/api.py
1
"""
6✔
2
Retrieve JSON schemas for validating dicts representing a ``pyproject.toml`` file.
3
"""
4

5
import json
6✔
6
import logging
6✔
7
import typing
6✔
8
from enum import Enum
6✔
9
from functools import partial, reduce
6✔
10
from types import MappingProxyType, ModuleType
6✔
11
from typing import (
6✔
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
6✔
24

25
from . import errors, formats
6✔
26
from .error_reporting import detailed_errors
6✔
27
from .extra_validations import EXTRA_VALIDATIONS
6✔
28
from .types import FormatValidationFn, Schema, ValidationFn
6✔
29

30
_logger = logging.getLogger(__name__)
6✔
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)
6✔
48
AllPlugins = Enum("AllPlugins", "ALL_PLUGINS")  #: :meta private:
6✔
49
ALL_PLUGINS = AllPlugins.ALL_PLUGINS
6✔
50

51
TOP_LEVEL_SCHEMA = "pyproject_toml"
6✔
52
PROJECT_TABLE_SCHEMA = "project_metadata"
6✔
53

54

55
def _get_public_functions(module: ModuleType) -> Mapping[str, FormatValidationFn]:
6✔
56
    return {
6✔
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))
6✔
64

65

66
def load(name: str, package: str = __package__, ext: str = ".schema.json") -> Schema:
6✔
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}")))
6✔
73

74

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

79

80
class SchemaRegistry(Mapping[str, Schema]):
6✔
81
    """Repository of parsed JSON Schemas used for validating a ``pyproject.toml``.
6✔
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"] = ()):
6✔
94
        self._schemas: Dict[str, Tuple[str, str, Schema]] = {}
6✔
95
        # (which part of the TOML, who defines, schema)
96

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

102
        # Add PEP 621
103
        project_table_schema = load(PROJECT_TABLE_SCHEMA)
6✔
104
        self._ensure_compatibility(PROJECT_TABLE_SCHEMA, project_table_schema)
6✔
105
        sid = project_table_schema["$id"]
6✔
106
        top_level["project"] = {"$ref": sid}
6✔
107
        origin = f"{__name__} - project metadata"
6✔
108
        self._schemas = {sid: ("project", origin, project_table_schema)}
6✔
109

110
        # Add tools using Plugins
111
        for plugin in plugins:
6✔
112
            allow_overwrite: Optional[str] = None
6✔
113
            if plugin.tool in tool_properties:
6✔
114
                _logger.warning(f"{plugin.id} overwrites `tool.{plugin.tool}` schema")
6✔
115
                allow_overwrite = plugin.schema.get("$id")
6✔
116
            else:
117
                _logger.info(f"{plugin.id} defines `tool.{plugin.tool}` schema")
6✔
118
            compatible = self._ensure_compatibility(
6✔
119
                plugin.tool, plugin.schema, allow_overwrite
120
            )
121
            sid = compatible["$id"]
6✔
122
            sref = f"{sid}#{plugin.fragment}" if plugin.fragment else sid
6✔
123
            tool_properties[plugin.tool] = {"$ref": sref}
6✔
124
            self._schemas[sid] = (f"tool.{plugin.tool}", plugin.id, plugin.schema)
6✔
125

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

131
    @property
6✔
132
    def spec_version(self) -> str:
6✔
133
        """Version of the JSON Schema spec in use"""
134
        return self._spec_version
6✔
135

136
    @property
6✔
137
    def main(self) -> str:
6✔
138
        """Top level schema for validating a ``pyproject.toml`` file"""
139
        return self._main_id
6✔
140

141
    def _ensure_compatibility(
6✔
142
        self, reference: str, schema: Schema, allow_overwrite: Optional[str] = None
143
    ) -> Schema:
144
        if "$id" not in schema or not schema["$id"]:
6✔
145
            raise errors.SchemaMissingId(reference)
6✔
146
        sid = schema["$id"]
6✔
147
        if sid in self._schemas and sid != allow_overwrite:
6✔
148
            raise errors.SchemaWithDuplicatedId(sid)
6✔
149
        version = schema.get("$schema")
6✔
150
        # Support schemas with missing trailing # (incorrect, but required before 0.15)
151
        if version and version.rstrip("#") != self.spec_version.rstrip("#"):
6✔
152
            raise errors.InvalidSchemaVersion(reference, version, self.spec_version)
6✔
153
        return schema
6✔
154

155
    def __getitem__(self, key: str) -> Schema:
6✔
156
        return self._schemas[key][-1]
6✔
157

158
    def __iter__(self) -> Iterator[str]:
6✔
159
        return iter(self._schemas)
×
160

161
    def __len__(self) -> int:
6✔
162
        return len(self._schemas)
×
163

164

165
class RefHandler(Mapping[str, Callable[[str], Schema]]):
6✔
166
    """:mod:`fastjsonschema` allows passing a dict-like object to load external schema
6✔
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]):
6✔
176
        self._uri_schemas = ["http", "https"]
6✔
177
        self._registry = registry
6✔
178

179
    def __contains__(self, key: object) -> bool:
6✔
180
        if isinstance(key, str):
6✔
181
            if key not in self._uri_schemas:
6✔
182
                self._uri_schemas.append(key)
×
183
            return True
6✔
184
        return False
×
185

186
    def __iter__(self) -> Iterator[str]:
6✔
187
        return iter(self._uri_schemas)
×
188

189
    def __len__(self) -> int:
6✔
190
        return len(self._uri_schemas)
×
191

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

196

197
class Validator:
6✔
198
    _plugins: Sequence["PluginProtocol"]
6✔
199

200
    def __init__(
6✔
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
6✔
209
        self._cache: Optional[ValidationFn] = None
6✔
210
        self._schema: Optional[Schema] = None
6✔
211

212
        # Let's make the following options readonly
213
        self._format_validators = MappingProxyType(format_validators)
6✔
214
        self._extra_validations = tuple(extra_validations)
6✔
215

216
        if plugins is ALL_PLUGINS:
6✔
217
            from .plugins import list_from_entry_points
6✔
218

219
            plugins = list_from_entry_points()
6✔
220

221
        self._plugins = (*plugins, *extra_plugins)
6✔
222

223
        self._schema_registry = SchemaRegistry(self._plugins)
6✔
224
        self.handlers = RefHandler(self._schema_registry)
6✔
225

226
    @property
6✔
227
    def registry(self) -> SchemaRegistry:
6✔
228
        return self._schema_registry
6✔
229

230
    @property
6✔
231
    def schema(self) -> Schema:
6✔
232
        """Top level ``pyproject.toml`` JSON Schema"""
233
        return Schema({"$ref": self._schema_registry.main})
6✔
234

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

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

245
    @property
6✔
246
    def generated_code(self) -> str:
6✔
247
        if self._code_cache is None:
6✔
248
            fmts = dict(self.formats)
6✔
249
            self._code_cache = FJS.compile_to_code(self.schema, self.handlers, fmts)
6✔
250

251
        return self._code_cache
6!
252

253
    def __getitem__(self, schema_id: str) -> Schema:
6✔
254
        """Retrieve a schema from registry"""
255
        return self._schema_registry[schema_id]
×
256

257
    def __call__(self, pyproject: T) -> T:
6✔
258
        """Checks a parsed ``pyproject.toml`` file (given as :obj:`typing.Mapping`)
259
        and raises an exception when it is not a valid.
260
        """
261
        if self._cache is None:
6✔
262
            compiled = FJS.compile(self.schema, self.handlers, dict(self.formats))
6✔
263
            fn = partial(compiled, custom_formats=self._format_validators)
6✔
264
            self._cache = typing.cast(ValidationFn, fn)
6✔
265

266
        with detailed_errors():
6✔
267
            self._cache(pyproject)
6✔
268
            return reduce(lambda acc, fn: fn(acc), self.extra_validations, pyproject)
6✔
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2025 Coveralls, Inc