# 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 collections import abc
from contextlib import contextmanager
from typing import Any
from typing import ClassVar
from typing import Iterable
from typing import Iterator
from typing import Mapping
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
class _MergeRequiredStrategy(Object):
"""A genson Object with a modified required attribute handling.
Genson Object does not merge the required attribute on purpose.
See :ref:`https://github.com/wolverdude/GenSON#genson`.
This class will merge the required attributes.
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, updated_names: Iterable[str], required: Iterable[str]
) -> None:
"""A context manager to handle the update vs merge.
Args:
updated_names: All the names to update.
required: The required names to add.
"""
# Pass the update switch to _SchemaNode.
self.node_class.update = self.update
if not self._required or not required:
yield
else:
# Backup the current required before updating it with the new ones.
_required = set(self._required)
yield
if self.update:
# The elements we update from overrule the existing ones, the required
# or not state for all the elements we update shall be reset.
_required -= updated_names
self._required = _required | set(required)
# 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:
"""Add a schema and merge the required attribute.
Args:
schema: A schema to be added.
"""
with self.__handle_update(
schema.get("properties", {}).keys(), schema.get("required")
):
super().add_schema(schema)
def add_object(self, obj: Mapping[str, Any]) -> None:
"""Add an object and merge the required attribute.
Args:
obj: An object to be added.
"""
with self.__handle_update(obj.keys(), obj.keys()):
super().add_object(obj)
class _SchemaNode(SchemaNode):
"""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) -> None:
self.__handle_update()
super().add_schema(schema)
def add_object(self, 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(type(abc.Mapping), _MetaSchemaBuilder):
"""Required meta class for inheriting from multiple classes with meta classes.
Also fix the ``NODE_CLASS`` overloading because it does not use the ``NODE_CLASS``
passed to a class derived from ``SchemaBuilder``.
"""
def __init__(self, name: str, bases: tuple(type), attrs: dict[str, Any]) -> None:
super().__init__(name, bases, attrs)
self.NODE_CLASS = type(
"%sSchemaNode" % name, (_SchemaNode,), {"STRATEGIES": self.STRATEGIES}
)
class _Number(Number):
"""A number strategy that handles numpy data."""
PYTHON_TYPES = Number.PYTHON_TYPES + (float64, int64)
[docs]class MutableMappingSchemaBuilder(abc.Mapping, SchemaBuilder, 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 = (_MergeRequiredStrategy, _Number)
NODE_CLASS = _SchemaNode
def __getitem__(self, key: str) -> dict[str, Any]:
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]
if key in self.required:
self.required.remove(key)
@property
def properties(self) -> dict[str, Any]:
"""Return the properties.
Returns:
The existing properties, otherwise an empty dictionary.
"""
try:
return 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 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:
raise KeyError(f"The name {name} is not in the grammar.")
[docs] def add_schema(self, schema, update: bool) -> None:
"""
Args:
update: Whether to update or merge the schema.
""" # noqa: D205 D212 D415
_MergeRequiredStrategy.update = update
super().add_schema(schema)
[docs] def add_object(self, obj, update: bool) -> None:
"""
Args:
update: Whether to update or merge the schema.
""" # noqa: D205 D212 D415
_MergeRequiredStrategy.update = update
super().add_object(obj)
