<commit_before>
#
# spyne - Copyright (C) Spyne contributors.
#
# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation; either
# version 2.1 of the License, or (at your option) any later version.
#
# This library 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 library; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301
#

import logging
logger = logging.getLogger(__name__)

import re
RE_HTTP_ARRAY_INDEX = re.compile("\\[([0-9]+)\\]")

from collections import defaultdict, Iterable as AbcIterable

from spyne.util import six
from spyne.error import ValidationError
from spyne.error import ResourceNotFoundError

from spyne.model import ByteArray, File, Fault, ComplexModelBase, Array, Any, \
    AnyDict, Uuid, Unicode

from spyne.protocol.dictdoc import DictDocument


class HierDictDocument(DictDocument):
    """This protocol contains logic for protocols that serialize and deserialize
    hierarchical dictionaries. Examples include: Json, MessagePack and Yaml.

    Implement ``create_in_document()`` and ``create_out_string()`` to use this.
    """

    from_serstr = DictDocument.from_unicode
    to_serstr = DictDocument.to_unicode

    def get_class_name(self, cls):
        class_name = cls.get_type_name()
        if not six.PY2:
            if isinstance(class_name, bytes):
                class_name = class_name.decode('utf8')

        return class_name

    def deserialize(self, ctx, message):
        assert message in (self.REQUEST, self.RESPONSE)

        self.event_manager.fire_event('before_deserialize', ctx)

        if ctx.descriptor is None:
            raise ResourceNotFoundError(ctx.method_request_string)

        # instantiate the result message
        if message is self.REQUEST:
            body_class = ctx.descriptor.in_message
        elif message is self.RESPONSE:
            body_class = ctx.descriptor.out_message

        if body_class:
            # assign raw result to its wrapper, result_message
            doc = ctx.in_body_doc

            class_name = self.get_class_name(body_class)
            if self.ignore_wrappers:
                doc = doc.get(class_name, None)
            result_message = self._doc_to_object(body_class, doc,
                                                                 self.validator)
            ctx.in_object = result_message

        else:
            ctx.in_object = []

        self.event_manager.fire_event('after_deserialize', ctx)

    def serialize(self, ctx, message):
        assert message in (self.REQUEST, self.RESPONSE)

        self.event_manager.fire_event('before_serialize', ctx)

        if ctx.out_error is not None:
            ctx.out_document = [Fault.to_dict(ctx.out_error.__class__,
                                                                 ctx.out_error)]
            return

        # get the result message
        if message is self.REQUEST:
            out_type = ctx.descriptor.in_message
        elif message is self.RESPONSE:
            out_type = ctx.descriptor.out_message
        if out_type is None:
            return

        out_type_info = out_type.get_flat_type_info(out_type)

        # instantiate the result message
        out_instance = out_type()

        # assign raw result to its wrapper, result_message
        for i, (k, v) in enumerate(out_type_info.items()):
            attr_name = k
            out_instance._safe_set(attr_name, ctx.out_object[i], v)

        ctx.out_document = self._object_to_doc(out_type, out_instance),

        self.event_manager.fire_event('after_serialize', ctx)

    def validate(self, key, cls, inst):
        if inst is None and self.get_cls_attrs(cls).nullable:
            pass

        elif issubclass(cls, Unicode) and not isinstance(inst,(six.text_type,
                                                               six.binary_type)):
            raise ValidationError((key, inst))

    def _from_dict_value(self, key, cls, inst, validator):
        if validator is self.SOFT_VALIDATION:
            self.validate(key, cls, inst)

        if issubclass(cls, (Any, AnyDict)):
            retval = inst

        # get native type
        elif issubclass(cls, File) and isinstance(inst, self.complex_as):
            retval = self._doc_to_object(cls.Attributes.type, inst, validator)

        elif issubclass(cls, ComplexModelBase):
            retval = self._doc_to_object(cls, inst, validator)

        else:
            if cls.Attributes.empty_is_none and inst in (u'', b''):
                inst = None

            if (validator is self.SOFT_VALIDATION
                                        and isinstance(inst, six.string_types)
                                        and not cls.validate_string(cls, inst)):
                raise ValidationError((key, inst))

            if issubclass(cls, (ByteArray, File, Uuid)):
                retval = self.from_serstr(cls, inst, self.binary_encoding)
            else:
                retval = self.from_serstr(cls, inst)

        # validate native type
        if validator is self.SOFT_VALIDATION and \
                                           not cls.validate_native(cls, retval):
            raise ValidationError((key, retval))

        return retval

    def _doc_to_object(self, cls, doc, validator=None):<commit_msg>Fix string interpolation funkiness with ValidationError.<commit_after>
#
# spyne - Copyright (C) Spyne contributors.
#
# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation; either
# version 2.1 of the License, or (at your option) any later version.
#
# This library 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 library; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301
#

import logging
logger = logging.getLogger(__name__)

import re
RE_HTTP_ARRAY_INDEX = re.compile("\\[([0-9]+)\\]")

from collections import defaultdict, Iterable as AbcIterable

from spyne.util import six
from spyne.error import ValidationError
from spyne.error import ResourceNotFoundError

from spyne.model import ByteArray, File, Fault, ComplexModelBase, Array, Any, \
    AnyDict, Uuid, Unicode

from spyne.protocol.dictdoc import DictDocument


class HierDictDocument(DictDocument):
    """This protocol contains logic for protocols that serialize and deserialize
    hierarchical dictionaries. Examples include: Json, MessagePack and Yaml.

    Implement ``create_in_document()`` and ``create_out_string()`` to use this.
    """

    from_serstr = DictDocument.from_unicode
    to_serstr = DictDocument.to_unicode

    def get_class_name(self, cls):
        class_name = cls.get_type_name()
        if not six.PY2:
            if isinstance(class_name, bytes):
                class_name = class_name.decode('utf8')

        return class_name

    def deserialize(self, ctx, message):
        assert message in (self.REQUEST, self.RESPONSE)

        self.event_manager.fire_event('before_deserialize', ctx)

        if ctx.descriptor is None:
            raise ResourceNotFoundError(ctx.method_request_string)

        # instantiate the result message
        if message is self.REQUEST:
            body_class = ctx.descriptor.in_message
        elif message is self.RESPONSE:
            body_class = ctx.descriptor.out_message

        if body_class:
            # assign raw result to its wrapper, result_message
            doc = ctx.in_body_doc

            class_name = self.get_class_name(body_class)
            if self.ignore_wrappers:
                doc = doc.get(class_name, None)
            result_message = self._doc_to_object(body_class, doc,
                                                                 self.validator)
            ctx.in_object = result_message

        else:
            ctx.in_object = []

        self.event_manager.fire_event('after_deserialize', ctx)

    def serialize(self, ctx, message):
        assert message in (self.REQUEST, self.RESPONSE)

        self.event_manager.fire_event('before_serialize', ctx)

        if ctx.out_error is not None:
            ctx.out_document = [Fault.to_dict(ctx.out_error.__class__,
                                                                 ctx.out_error)]
            return

        # get the result message
        if message is self.REQUEST:
            out_type = ctx.descriptor.in_message
        elif message is self.RESPONSE:
            out_type = ctx.descriptor.out_message
        if out_type is None:
            return

        out_type_info = out_type.get_flat_type_info(out_type)

        # instantiate the result message
        out_instance = out_type()

        # assign raw result to its wrapper, result_message
        for i, (k, v) in enumerate(out_type_info.items()):
            attr_name = k
            out_instance._safe_set(attr_name, ctx.out_object[i], v)

        ctx.out_document = self._object_to_doc(out_type, out_instance),

        self.event_manager.fire_event('after_serialize', ctx)

    def validate(self, key, cls, inst):
        if inst is None and self.get_cls_attrs(cls).nullable:
            pass

        elif issubclass(cls, Unicode) and not isinstance(inst,(six.text_type,
                                                               six.binary_type)):
            raise ValidationError((key, inst))

    def _from_dict_value(self, key, cls, inst, validator):
        if validator is self.SOFT_VALIDATION:
            self.validate(key, cls, inst)

        if issubclass(cls, (Any, AnyDict)):
            retval = inst

        # get native type
        elif issubclass(cls, File) and isinstance(inst, self.complex_as):
            retval = self._doc_to_object(cls.Attributes.type, inst, validator)

        elif issubclass(cls, ComplexModelBase):
            retval = self._doc_to_object(cls, inst, validator)

        else:
            if cls.Attributes.empty_is_none and inst in (u'', b''):
                inst = None

            if (validator is self.SOFT_VALIDATION
                                        and isinstance(inst, six.string_types)
                                        and not cls.validate_string(cls, inst)):
                raise ValidationError((key, inst))

            if issubclass(cls, (ByteArray, File, Uuid)):
                retval = self.from_serstr(cls, inst, self.binary_encoding)
            else:
                retval = self.from_serstr(cls, inst)

        # validate native type
        if validator is self.SOFT_VALIDATION and \
                                           not cls.validate_native(cls, retval):
            raise ValidationError([key, retval])

        return retval

    def _doc_to_object(self, cls, doc, validator=None):
{'fraction_non_alphanumeric': 0.07579845096162213, 'fraction_numerical': 0.0029588373509703246, 'mean_word_length': 2.245410900875459, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 24, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2989324', 'n_tokens_mistral': 3133, 'n_tokens_neox': 3042, 'n_words': 928}