codekingpro/portable-devtools

1"""2Referencing implementations for JSON Schema specs (historic & current).3"""4 5from __future__ import annotations6 7from collections.abc import Iterable, Sequence, Set8from typing import Any9 10from referencing import Anchor, Registry, Resource, Specification, exceptions11from referencing._attrs import frozen12from referencing._core import (13    _UNSET,  # type: ignore[reportPrivateUsage]14    Resolved as _Resolved,15    Resolver as _Resolver,16    _Unset,  # type: ignore[reportPrivateUsage]17)18from referencing.typing import URI, Anchor as AnchorType, Mapping19 20#: A JSON Schema which is a JSON object21ObjectSchema = Mapping[str, Any]22 23#: A JSON Schema of any kind24Schema = bool | ObjectSchema25 26#: A Resource whose contents are JSON Schemas27SchemaResource = Resource[Schema]28 29#: A JSON Schema Registry30SchemaRegistry = Registry[Schema]31 32#: The empty JSON Schema Registry33EMPTY_REGISTRY: SchemaRegistry = Registry()34 35 36@frozen37class UnknownDialect(Exception):38    """39    A dialect identifier was found for a dialect unknown by this library.40 41    If it's a custom ("unofficial") dialect, be sure you've registered it.42    """43 44    uri: URI45 46 47def _dollar_id(contents: Schema) -> URI | None:48    if isinstance(contents, bool):49        return50    return contents.get("$id")51 52 53def _legacy_dollar_id(contents: Schema) -> URI | None:54    if isinstance(contents, bool) or "$ref" in contents:55        return56    id = contents.get("$id")57    if id is not None and not id.startswith("#"):58        return id59 60 61def _legacy_id(contents: ObjectSchema) -> URI | None:62    if "$ref" in contents:63        return64    id = contents.get("id")65    if id is not None and not id.startswith("#"):66        return id67 68 69def _anchor(70    specification: Specification[Schema],71    contents: Schema,72) -> Iterable[AnchorType[Schema]]:73    if isinstance(contents, bool):74        return75    anchor = contents.get("$anchor")76    if anchor is not None:77        yield Anchor(78            name=anchor,79            resource=specification.create_resource(contents),80        )81 82    dynamic_anchor = contents.get("$dynamicAnchor")83    if dynamic_anchor is not None:84        yield DynamicAnchor(85            name=dynamic_anchor,86            resource=specification.create_resource(contents),87        )88 89 90def _anchor_2019(91    specification: Specification[Schema],92    contents: Schema,93) -> Iterable[Anchor[Schema]]:94    if isinstance(contents, bool):95        return []96    anchor = contents.get("$anchor")97    if anchor is None:98        return []99    return [100        Anchor(101            name=anchor,102            resource=specification.create_resource(contents),103        ),104    ]105 106 107def _legacy_anchor_in_dollar_id(108    specification: Specification[Schema],109    contents: Schema,110) -> Iterable[Anchor[Schema]]:111    if isinstance(contents, bool):112        return []113    id = contents.get("$id", "")114    if not id.startswith("#"):115        return []116    return [117        Anchor(118            name=id[1:],119            resource=specification.create_resource(contents),120        ),121    ]122 123 124def _legacy_anchor_in_id(125    specification: Specification[ObjectSchema],126    contents: ObjectSchema,127) -> Iterable[Anchor[ObjectSchema]]:128    id = contents.get("id", "")129    if not id.startswith("#"):130        return []131    return [132        Anchor(133            name=id[1:],134            resource=specification.create_resource(contents),135        ),136    ]137 138 139def _subresources_of(140    in_value: Set[str] = frozenset(),141    in_subvalues: Set[str] = frozenset(),142    in_subarray: Set[str] = frozenset(),143):144    """145    Create a callable returning JSON Schema specification-style subschemas.146 147    Relies on specifying the set of keywords containing subschemas in their148    values, in a subobject's values, or in a subarray.149    """150 151    def subresources_of(contents: Schema) -> Iterable[ObjectSchema]:152        if isinstance(contents, bool):153            return154        for each in in_value:155            if each in contents:156                yield contents[each]157        for each in in_subarray:158            if each in contents:159                yield from contents[each]160        for each in in_subvalues:161            if each in contents:162                yield from contents[each].values()163 164    return subresources_of165 166 167def _subresources_of_with_crazy_items(168    in_value: Set[str] = frozenset(),169    in_subvalues: Set[str] = frozenset(),170    in_subarray: Set[str] = frozenset(),171):172    """173    Specifically handle older drafts where there are some funky keywords.174    """175 176    def subresources_of(contents: Schema) -> Iterable[ObjectSchema]:177        if isinstance(contents, bool):178            return179        for each in in_value:180            if each in contents:181                yield contents[each]182        for each in in_subarray:183            if each in contents:184                yield from contents[each]185        for each in in_subvalues:186            if each in contents:187                yield from contents[each].values()188 189        items = contents.get("items")190        if items is not None:191            if isinstance(items, Sequence):192                yield from items193            else:194                yield items195 196    return subresources_of197 198 199def _subresources_of_with_crazy_items_dependencies(200    in_value: Set[str] = frozenset(),201    in_subvalues: Set[str] = frozenset(),202    in_subarray: Set[str] = frozenset(),203):204    """205    Specifically handle older drafts where there are some funky keywords.206    """207 208    def subresources_of(contents: Schema) -> Iterable[ObjectSchema]:209        if isinstance(contents, bool):210            return211        for each in in_value:212            if each in contents:213                yield contents[each]214        for each in in_subarray:215            if each in contents:216                yield from contents[each]217        for each in in_subvalues:218            if each in contents:219                yield from contents[each].values()220 221        items = contents.get("items")222        if items is not None:223            if isinstance(items, Sequence):224                yield from items225            else:226                yield items227        dependencies = contents.get("dependencies")228        if dependencies is not None:229            values = iter(dependencies.values())230            value = next(values, None)231            if isinstance(value, Mapping):232                yield value233                yield from values234 235    return subresources_of236 237 238def _subresources_of_with_crazy_aP_items_dependencies(239    in_value: Set[str] = frozenset(),240    in_subvalues: Set[str] = frozenset(),241    in_subarray: Set[str] = frozenset(),242):243    """244    Specifically handle even older drafts where there are some funky keywords.245    """246 247    def subresources_of(contents: ObjectSchema) -> Iterable[ObjectSchema]:248        for each in in_value:249            if each in contents:250                yield contents[each]251        for each in in_subarray:252            if each in contents:253                yield from contents[each]254        for each in in_subvalues:255            if each in contents:256                yield from contents[each].values()257 258        items = contents.get("items")259        if items is not None:260            if isinstance(items, Sequence):261                yield from items262            else:263                yield items264        dependencies = contents.get("dependencies")265        if dependencies is not None:266            values = iter(dependencies.values())267            value = next(values, None)268            if isinstance(value, Mapping):269                yield value270                yield from values271 272        for each in "additionalItems", "additionalProperties":273            value = contents.get(each)274            if isinstance(value, Mapping):275                yield value276 277    return subresources_of278 279 280def _maybe_in_subresource(281    in_value: Set[str] = frozenset(),282    in_subvalues: Set[str] = frozenset(),283    in_subarray: Set[str] = frozenset(),284):285    in_child = in_subvalues | in_subarray286 287    def maybe_in_subresource(288        segments: Sequence[int | str],289        resolver: _Resolver[Any],290        subresource: Resource[Any],291    ) -> _Resolver[Any]:292        _segments = iter(segments)293        for segment in _segments:294            if segment not in in_value and (295                segment not in in_child or next(_segments, None) is None296            ):297                return resolver298        return resolver.in_subresource(subresource)299 300    return maybe_in_subresource301 302 303def _maybe_in_subresource_crazy_items(304    in_value: Set[str] = frozenset(),305    in_subvalues: Set[str] = frozenset(),306    in_subarray: Set[str] = frozenset(),307):308    in_child = in_subvalues | in_subarray309 310    def maybe_in_subresource(311        segments: Sequence[int | str],312        resolver: _Resolver[Any],313        subresource: Resource[Any],314    ) -> _Resolver[Any]:315        _segments = iter(segments)316        for segment in _segments:317            if segment == "items" and isinstance(318                subresource.contents,319                Mapping,320            ):321                return resolver.in_subresource(subresource)322            if segment not in in_value and (323                segment not in in_child or next(_segments, None) is None324            ):325                return resolver326        return resolver.in_subresource(subresource)327 328    return maybe_in_subresource329 330 331def _maybe_in_subresource_crazy_items_dependencies(332    in_value: Set[str] = frozenset(),333    in_subvalues: Set[str] = frozenset(),334    in_subarray: Set[str] = frozenset(),335):336    in_child = in_subvalues | in_subarray337 338    def maybe_in_subresource(339        segments: Sequence[int | str],340        resolver: _Resolver[Any],341        subresource: Resource[Any],342    ) -> _Resolver[Any]:343        _segments = iter(segments)344        for segment in _segments:345            if segment in {"items", "dependencies"} and isinstance(346                subresource.contents,347                Mapping,348            ):349                return resolver.in_subresource(subresource)350            if segment not in in_value and (351                segment not in in_child or next(_segments, None) is None352            ):353                return resolver354        return resolver.in_subresource(subresource)355 356    return maybe_in_subresource357 358 359#: JSON Schema draft 2020-12360DRAFT202012 = Specification(361    name="draft2020-12",362    id_of=_dollar_id,363    subresources_of=_subresources_of(364        in_value={365            "additionalProperties",366            "contains",367            "contentSchema",368            "else",369            "if",370            "items",371            "not",372            "propertyNames",373            "then",374            "unevaluatedItems",375            "unevaluatedProperties",376        },377        in_subarray={"allOf", "anyOf", "oneOf", "prefixItems"},378        in_subvalues={379            "$defs",380            "definitions",381            "dependentSchemas",382            "patternProperties",383            "properties",384        },385    ),386    anchors_in=_anchor,387    maybe_in_subresource=_maybe_in_subresource(388        in_value={389            "additionalProperties",390            "contains",391            "contentSchema",392            "else",393            "if",394            "items",395            "not",396            "propertyNames",397            "then",398            "unevaluatedItems",399            "unevaluatedProperties",400        },401        in_subarray={"allOf", "anyOf", "oneOf", "prefixItems"},402        in_subvalues={403            "$defs",404            "definitions",405            "dependentSchemas",406            "patternProperties",407            "properties",408        },409    ),410)411#: JSON Schema draft 2019-09412DRAFT201909 = Specification(413    name="draft2019-09",414    id_of=_dollar_id,415    subresources_of=_subresources_of_with_crazy_items(416        in_value={417            "additionalItems",418            "additionalProperties",419            "contains",420            "contentSchema",421            "else",422            "if",423            "not",424            "propertyNames",425            "then",426            "unevaluatedItems",427            "unevaluatedProperties",428        },429        in_subarray={"allOf", "anyOf", "oneOf"},430        in_subvalues={431            "$defs",432            "definitions",433            "dependentSchemas",434            "patternProperties",435            "properties",436        },437    ),438    anchors_in=_anchor_2019,439    maybe_in_subresource=_maybe_in_subresource_crazy_items(440        in_value={441            "additionalItems",442            "additionalProperties",443            "contains",444            "contentSchema",445            "else",446            "if",447            "not",448            "propertyNames",449            "then",450            "unevaluatedItems",451            "unevaluatedProperties",452        },453        in_subarray={"allOf", "anyOf", "oneOf"},454        in_subvalues={455            "$defs",456            "definitions",457            "dependentSchemas",458            "patternProperties",459            "properties",460        },461    ),462)463#: JSON Schema draft 7464DRAFT7 = Specification(465    name="draft-07",466    id_of=_legacy_dollar_id,467    subresources_of=_subresources_of_with_crazy_items_dependencies(468        in_value={469            "additionalItems",470            "additionalProperties",471            "contains",472            "else",473            "if",474            "not",475            "propertyNames",476            "then",477        },478        in_subarray={"allOf", "anyOf", "oneOf"},479        in_subvalues={"definitions", "patternProperties", "properties"},480    ),481    anchors_in=_legacy_anchor_in_dollar_id,482    maybe_in_subresource=_maybe_in_subresource_crazy_items_dependencies(483        in_value={484            "additionalItems",485            "additionalProperties",486            "contains",487            "else",488            "if",489            "not",490            "propertyNames",491            "then",492        },493        in_subarray={"allOf", "anyOf", "oneOf"},494        in_subvalues={"definitions", "patternProperties", "properties"},495    ),496)497#: JSON Schema draft 6498DRAFT6 = Specification(499    name="draft-06",500    id_of=_legacy_dollar_id,501    subresources_of=_subresources_of_with_crazy_items_dependencies(502        in_value={503            "additionalItems",504            "additionalProperties",505            "contains",506            "not",507            "propertyNames",508        },509        in_subarray={"allOf", "anyOf", "oneOf"},510        in_subvalues={"definitions", "patternProperties", "properties"},511    ),512    anchors_in=_legacy_anchor_in_dollar_id,513    maybe_in_subresource=_maybe_in_subresource_crazy_items_dependencies(514        in_value={515            "additionalItems",516            "additionalProperties",517            "contains",518            "not",519            "propertyNames",520        },521        in_subarray={"allOf", "anyOf", "oneOf"},522        in_subvalues={"definitions", "patternProperties", "properties"},523    ),524)525#: JSON Schema draft 4526DRAFT4 = Specification(527    name="draft-04",528    id_of=_legacy_id,529    subresources_of=_subresources_of_with_crazy_aP_items_dependencies(530        in_value={"not"},531        in_subarray={"allOf", "anyOf", "oneOf"},532        in_subvalues={"definitions", "patternProperties", "properties"},533    ),534    anchors_in=_legacy_anchor_in_id,535    maybe_in_subresource=_maybe_in_subresource_crazy_items_dependencies(536        in_value={"additionalItems", "additionalProperties", "not"},537        in_subarray={"allOf", "anyOf", "oneOf"},538        in_subvalues={"definitions", "patternProperties", "properties"},539    ),540)541#: JSON Schema draft 3542DRAFT3 = Specification(543    name="draft-03",544    id_of=_legacy_id,545    subresources_of=_subresources_of_with_crazy_aP_items_dependencies(546        in_subarray={"extends"},547        in_subvalues={"definitions", "patternProperties", "properties"},548    ),549    anchors_in=_legacy_anchor_in_id,550    maybe_in_subresource=_maybe_in_subresource_crazy_items_dependencies(551        in_value={"additionalItems", "additionalProperties"},552        in_subarray={"extends"},553        in_subvalues={"definitions", "patternProperties", "properties"},554    ),555)556 557 558_SPECIFICATIONS: Registry[Specification[Schema]] = Registry(559    {560        dialect_id: Resource.opaque(specification)561        for dialect_id, specification in [562            ("https://json-schema.org/draft/2020-12/schema", DRAFT202012),563            ("https://json-schema.org/draft/2019-09/schema", DRAFT201909),564            ("http://json-schema.org/draft-07/schema", DRAFT7),565            ("http://json-schema.org/draft-06/schema", DRAFT6),566            ("http://json-schema.org/draft-04/schema", DRAFT4),567            ("http://json-schema.org/draft-03/schema", DRAFT3),568        ]569    },570)571 572 573def specification_with(574    dialect_id: URI,575    default: Specification[Any] | _Unset = _UNSET,576) -> Specification[Any]:577    """578    Retrieve the `Specification` with the given dialect identifier.579 580    Raises:581 582        `UnknownDialect`583 584            if the given ``dialect_id`` isn't known585 586    """587    resource = _SPECIFICATIONS.get(dialect_id.rstrip("#"))588    if resource is not None:589        return resource.contents590    if default is _UNSET:591        raise UnknownDialect(dialect_id)592    return default593 594 595@frozen596class DynamicAnchor:597    """598    Dynamic anchors, introduced in draft 2020.599    """600 601    name: str602    resource: SchemaResource603 604    def resolve(self, resolver: _Resolver[Schema]) -> _Resolved[Schema]:605        """606        Resolve this anchor dynamically.607        """608        last = self.resource609        for uri, registry in resolver.dynamic_scope():610            try:611                anchor = registry.anchor(uri, self.name).value612            except exceptions.NoSuchAnchor:613                continue614            if isinstance(anchor, DynamicAnchor):615                last = anchor.resource616        return _Resolved(617            contents=last.contents,618            resolver=resolver.in_subresource(last),619        )620 621 622def lookup_recursive_ref(resolver: _Resolver[Schema]) -> _Resolved[Schema]:623    """624    Recursive references (via recursive anchors), present only in draft 2019.625 626    As per the 2019 specification (§ 8.2.4.2.1), only the ``#`` recursive627    reference is supported (and is therefore assumed to be the relevant628    reference).629    """630    resolved = resolver.lookup("#")631    if isinstance(resolved.contents, Mapping) and resolved.contents.get(632        "$recursiveAnchor",633    ):634        for uri, _ in resolver.dynamic_scope():635            next_resolved = resolver.lookup(uri)636            if not isinstance(637                next_resolved.contents,638                Mapping,639            ) or not next_resolved.contents.get("$recursiveAnchor"):640                break641            resolved = next_resolved642    return resolved643