model.py

##############################################################################
#
#    Converter Odoo module
#    Copyright © 2020, 2024 XCG Consulting <https://xcg-consulting.fr>
#
#    This program is free software: you can redistribute it and/or modify
#    it under the terms of the GNU Affero General Public License as
#    published by the Free Software Foundation, either version 3 of the
#    License, or (at your option) any later version.
#
#    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 Affero General Public License for more details.
#
#    You should have received a copy of the GNU Affero General Public License
#    along with this program.  If not, see <http://www.gnu.org/licenses/>.
#
##############################################################################
import logging
import traceback
from collections import OrderedDict
from collections.abc import Iterable, Mapping
from typing import Any, Final

from odoo import _, api, models  # type: ignore[import-untyped]
from odoo.exceptions import UserError  # type: ignore[import-untyped]

from .base import (
    Context,
    ContextBuilder,
    Converter,
    Newinstance,
    NewinstanceType,
    PostHookConverter,
    SkipType,
    build_context,
)
from .validate import NotInitialized, Validation, Validator

try:
    from fastjsonschema import JsonSchemaException  # type: ignore[import-untyped]
except ImportError:
    # Ignore no-redef, added for compatibility
    class JsonSchemaException(Exception):  # type: ignore[no-redef]
        """Custom error in case of missing optional requirement"""


_logger = logging.getLogger(__name__)


class IncorrectTypeException(Exception):
    """__type__ is in the message is not the same as the expected value"""


class MissingRequiredValidatorException(Exception):
    def __init__(self):
        super().__init__("Strict validation without validator")


class Model(PostHookConverter):
    """A converter that takes a dict of key, used when a message has values"""

    def __init__(
        self,
        converters: Mapping[str, Converter],
        json_schema: str | None = None,
        # The validator is usually not given at this point but is common
        # throughout a project. That’s why it is a property
        validator: Validator | None = None,
        merge_with: Iterable[Converter] | None = None,
        validation: Validation = Validation.SKIP,
        context: ContextBuilder | None = None,
        datatype: str | None = None,
        __type__: str | None = None,
    ):
        """
        :param datatype: datatype to use. Usually used with None __type__.
        """
        super().__init__()
        self._type: str | None = __type__
        self._converters: Mapping[str, Converter] = converters
        self._post_hooks_converters: dict[str, PostHookConverter] = {}
        self._jsonschema: str | None = json_schema
        self._get_instance: str | None = None
        """First converter whose `is_instance_getter` is true if any"""
        self.merge_with: Iterable[Converter] | None = merge_with
        self.context: ContextBuilder | None = context
        self.validator = validator
        self.validation = validation
        self._datatype: Final[str | None] = datatype

        for key, converter in converters.items():
            if self._get_instance is None and converter.is_instance_getter:
                self._get_instance = key
            if isinstance(converter, PostHookConverter):
                self._post_hooks_converters[key] = converter

    def odoo_to_message(self, instance: models.BaseModel, ctx: Context = None) -> Any:
        ctx = build_context(instance, ctx, self.context)

        message_data = {}
        if self._type is not None:
            message_data["__type__"] = self._type

        errors = []
        for key in self._converters:
            try:
                value = self._converters[key].odoo_to_message(instance, ctx)
            except Exception as e:
                errors.append(
                    {"key": key, "traceback": "".join(traceback.format_exception(e))}
                )
                continue
            if not isinstance(value, SkipType):
                message_data[key] = value
        if len(errors) != 0:
            formatted_errors = "\n\n".join(
                [f"{error['traceback']}Key: {error['key']}" for error in errors]
            )
            raise UserError(
                _(
                    "Got unexpected errors while parsing substitutions:\n%s",
                    formatted_errors,
                )
            )

        if self.merge_with:
            for conv in self.merge_with:
                value = conv.odoo_to_message(instance, ctx)
                if isinstance(value, SkipType):
                    continue
                message_data.update(value)

        if self.validation != Validation.SKIP and self._jsonschema is not None:
            if self.validator:
                try:
                    self.validator.validate(self._jsonschema, message_data)
                except (NotInitialized, JsonSchemaException):
                    _logger.warning("Validation failed", exc_info=True)
                    if self.validation == Validation.STRICT:
                        raise
            elif self.validation == Validation.STRICT:
                raise MissingRequiredValidatorException()

        return message_data

    def message_to_odoo(
        self,
        odoo_env: api.Environment,
        phase: str,
        message_value: Any,
        instance: models.BaseModel,
        value_present: bool = True,
    ) -> dict | SkipType:
        values: dict[str, Any] = OrderedDict()

        if self.validation != Validation.SKIP and self._jsonschema is not None:
            if self.validator:
                try:
                    self.validator.validate(self._jsonschema, message_value)
                except (NotInitialized, JsonSchemaException):
                    _logger.warning("Validation failed", exc_info=True)
                    if self.validation == Validation.STRICT:
                        raise
            elif self.validation == Validation.STRICT:
                raise MissingRequiredValidatorException()

        if self._type is not None and message_value["__type__"] != self._type:
            raise IncorrectTypeException(
                "Expected __type__ {}, found {}".format(
                    self._type, message_value["__type__"]
                )
            )
        for key in self._converters:
            value = message_value.get(key, None) if message_value else None
            attribute_vals = self._converters[key].message_to_odoo(
                odoo_env,
                phase,
                value,
                instance,
                message_value and key in message_value,
            )
            if isinstance(attribute_vals, SkipType):
                continue
            values.update(attribute_vals)
        if self.merge_with:
            for conv in self.merge_with:
                value = conv.message_to_odoo(
                    odoo_env, phase, message_value, instance, value_present
                )
                if isinstance(value, SkipType):
                    continue
                values.update(value)

        return values

    @property
    def is_instance_getter(self) -> bool:
        return self._get_instance is not None

    def get_instance(
        self, odoo_env: api.Environment, message_data
    ) -> models.BaseModel | NewinstanceType | None:
        """:return: an instance linked to the converter, if any"""
        if self._get_instance:
            instance = self._converters[self._get_instance].get_instance(
                odoo_env, message_data[self._get_instance]
            )
            if instance is None:
                instance = Newinstance
            return instance
        return None

    def post_hook(self, instance: models.BaseModel, message_data):
        for key in self._post_hooks_converters:
            if key in message_data:
                self._post_hooks_converters[key].post_hook(instance, message_data[key])
        if self.merge_with:
            for converter in self.merge_with:
                if hasattr(converter, "post_hook"):
                    converter.post_hook(instance, message_data)

    def get__type__(self) -> set[str]:
        return set() if self._type is None else {self._type}

    @property
    def possible_datatypes(self) -> set[str]:
        result = set()
        if self._datatype is not None:
            result.add(self._datatype)
        return result

    def odoo_datatype(self, instance: models.BaseModel) -> str | None:
        return self._datatype