codekingpro/portable-devtools

1"""Utilities for storing collections of error messages.2 3.. warning::4 5    This module is treated as private API.6    Users should not need to use this module directly.7"""8 9from marshmallow.exceptions import SCHEMA10 11 12class ErrorStore:13    def __init__(self):14        #: Dictionary of errors stored during serialization15        self.errors = {}16 17    def store_error(self, messages, field_name=SCHEMA, index=None):18        # field error  -> store/merge error messages under field name key19        # schema error -> if string or list, store/merge under _schema key20        #              -> if dict, store/merge with other top-level keys21        messages = copy_containers(messages)22        if field_name != SCHEMA or not isinstance(messages, dict):23            messages = {field_name: messages}24        if index is not None:25            messages = {index: messages}26        self.errors = merge_errors(self.errors, messages)27 28 29def copy_containers(errors):30    if isinstance(errors, list):31        return [copy_containers(val) for val in errors]32    if isinstance(errors, dict):33        return {key: copy_containers(val) for key, val in errors.items()}34    return errors35 36 37def merge_errors(errors1, errors2):  # noqa: PLR091138    """Deeply merge two error messages.39 40    The format of ``errors1`` and ``errors2`` matches the ``message``41    parameter of :exc:`marshmallow.exceptions.ValidationError`.42    """43    if not errors1:44        return errors245    if not errors2:46        return errors147    if isinstance(errors1, list):48        if isinstance(errors2, list):49            errors1.extend(errors2)50            return errors151        if isinstance(errors2, dict):52            errors2[SCHEMA] = merge_errors(errors1, errors2.get(SCHEMA))53            return errors254        errors1.append(errors2)55        return errors156    if isinstance(errors1, dict):57        if isinstance(errors2, dict):58            for key, val in errors2.items():59                if key in errors1:60                    errors1[key] = merge_errors(errors1[key], val)61                else:62                    errors1[key] = val63            return errors164        errors1[SCHEMA] = merge_errors(errors1.get(SCHEMA), errors2)65        return errors166    if isinstance(errors2, list):67        return [errors1, *errors2]68    if isinstance(errors2, dict):69        errors2[SCHEMA] = merge_errors(errors1, errors2.get(SCHEMA))70        return errors271    return [errors1, errors2]72