Source code for gemseo.core.grammars.json_schema

# Copyright 2021 IRT Saint Exupéry, https://www.irt-saintexupery.com
#
# This program is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License version 3 as published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this program; if not, write to the Free Software Foundation,
# Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
"""JSON schema handler."""

from __future__ import annotations

from abc import ABCMeta
from collections import abc
from collections.abc import MutableMapping
from contextlib import contextmanager
from typing import TYPE_CHECKING
from typing import Any
from typing import ClassVar
from typing import TypedDict
from typing import cast

from genson import SchemaBuilder
from genson import SchemaNode
from genson.schema.builder import _MetaSchemaBuilder
from genson.schema.strategies import Number
from genson.schema.strategies import Object
from numpy import float64
from numpy import int64

if TYPE_CHECKING:
    from collections.abc import Iterator
    from collections.abc import Mapping

    from typing_extensions import NotRequired

    Property = TypedDict(  # noqa: UP013
        "Property",
        {"type": str, "items": NotRequired["Property"]},
        total=False,
    )

    Properties = dict[str, Property]

    Schema = TypedDict(  # noqa: UP013
        "Schema",
        {
            "properties": Properties,
            "type": str,
            "id": NotRequired[str],
            "required": NotRequired[list[str]],
            "$schema": str,
        },
        total=False,
    )

    Obj = Mapping[str, Any]

SchemaBuilderProperties = MutableMapping[str, SchemaNode]


class _MergeStrategy(Object):  # type: ignore
    """A genson strategy to either merge or update a schema.

    By default, genson merges nodes, the update is triggered via a class attribute.
    """

    # Do not merge the name and id properties.
    KEYWORDS = (*Object.KEYWORDS, "name", "id")

    update: ClassVar[bool] = False
    """Whether to update or merge the schema."""

    @contextmanager
    def __handle_update(self) -> Iterator[None]:
        """A context manager to handle the update vs merge."""
        # Pass the update switch to _SchemaNode.
        self.node_class.update = self.update
        yield
        # Reset to the merge behavior because _SchemaNode may be used by other instances
        # that should merge.
        self.node_class.update = False

    def add_schema(self, schema: Mapping[str, Any]) -> None:
        with self.__handle_update():
            super().add_schema(schema)

    def add_object(self, obj: Mapping[str, Any]) -> None:
        with self.__handle_update():
            super().add_object(obj)


class _SchemaNode(SchemaNode):  # type: ignore
    """Overload :meth:`.add_schema` and :meth:`.add_object` to allow updating.

    By default, genson merges nodes, the update is triggered via a class attribute.
    """

    update: ClassVar[bool] = False
    """Whether to update or merge the schema."""

    def add_schema(self, schema: Mapping[str, Any]) -> None:
        self.__handle_update()
        super().add_schema(schema)

    def add_object(self, obj: Obj) -> None:
        self.__handle_update()
        super().add_object(obj)

    def __handle_update(self) -> None:
        """Handle the update or merge behavior.

        When updating, the already existing active strategies are removed such that only
        the last one added remains.
        """
        if self.update and self._active_strategies:
            self._active_strategies.clear()


class _MultipleMeta(ABCMeta, _MetaSchemaBuilder):  # type: ignore
    """Required metaclass for inheriting from multiple classes with metaclasses.

    Also fix the ``NODE_CLASS`` overloading because it does not use the ``NODE_CLASS``
    passed to a class derived from ``SchemaBuilder``.
    """

    def __init__(cls, name: str, bases: tuple[type], attrs: dict[str, Any]) -> None:
        super().__init__(name, bases, attrs)
        cls.NODE_CLASS = type(
            "%sSchemaNode" % name, (_SchemaNode,), {"STRATEGIES": cls.STRATEGIES}
        )


class _Number(Number):  # type: ignore
    """A number strategy that handles numpy data."""

    PYTHON_TYPES = (*Number.PYTHON_TYPES, float64, int64)


[docs] class MutableMappingSchemaBuilder( abc.Mapping[str, Any], SchemaBuilder, # type: ignore metaclass=_MultipleMeta, ): """A mutable genson SchemaBuilder with a dictionary-like interface. The :class:`SchemaBuilder` does not provide a way to mutate directly the properties of a schema (these are stored deeply). For ease of usage, this class brings the properties closer to the surface, and the mutability is only provided by the ability to delete a property. """ EXTRA_STRATEGIES = (_MergeStrategy, _Number) NODE_CLASS = _SchemaNode def __getitem__(self, key: str) -> SchemaNode: self.check_property_names(key) return self.properties[key] def __iter__(self) -> Iterator[str]: return iter(self.properties) def __len__(self) -> int: return len(self.properties) def __delitem__(self, key: str) -> None: del self.properties[key] @property def properties(self) -> SchemaBuilderProperties: """Return the properties. Returns: The existing properties, otherwise an empty dictionary. """ try: return cast( SchemaBuilderProperties, self._root_node._active_strategies[0]._properties, ) except (AttributeError, IndexError): return {} @property def required(self) -> set[str]: """Return the required properties. Returns: The required properties, otherwise an empty set. """ try: required = self._root_node._active_strategies[0]._required except (AttributeError, IndexError): return set() if required is None: return set() return cast(set[str], required)
[docs] def check_property_names(self, *names: str) -> None: """Check that the names are existing properties. Args: *names: The names to be checked. Raises: KeyError: If a name is not an existing property. """ for name in names: if name not in self.properties: msg = f"The name {name} is not in the grammar." raise KeyError(msg)
[docs] def add_schema(self, schema: Schema, update: bool) -> None: """ Args: update: Whether to update or merge the schema. """ # noqa: D205 D212 D415 _MergeStrategy.update = update super().add_schema(schema)
[docs] def add_object(self, obj: Obj, update: bool) -> None: """ Args: update: Whether to update or merge the schema. """ # noqa: D205 D212 D415 _MergeStrategy.update = update super().add_object(obj)