def schema_merge()

in src/rpdk/core/jsonutils/utils.py [0:0]

• 44 lines of code
• 20 McCabe index (conditional complexity)

def schema_merge(target, src, path):  # noqa: C901 # pylint: disable=R0912
    """Merges the src schema into the target schema in place.

    If there are duplicate keys, src will overwrite target.

    :raises TypeError: either schema is not of type dict
    :raises ConstraintError: the schema tries to override "type" or "$ref"

    >>> schema_merge({}, {}, ())
    {}
    >>> schema_merge({'foo': 'a'}, {}, ())
    {'foo': 'a'}

    >>> schema_merge({}, {'foo': 'a'}, ())
    {'foo': 'a'}

    >>> schema_merge({'foo': 'a'}, {'foo': 'b'}, ())
    {'foo': 'b'}

    >>> schema_merge({'required': 'a'}, {'required': 'b'}, ())
    {'required': ['a', 'b']}

    >>> a, b = {'$ref': 'a'}, {'foo': 'b'}
    >>> schema_merge(a, b, ('foo',))
    {'$ref': 'a', 'foo': 'b'}

    >>> a, b = {'$ref': 'a'}, {'type': 'b'}
    >>> schema_merge(a, b, ('foo',))
    {'type': OrderedSet(['a', 'b'])}

    >>> a, b = {'$ref': 'a'}, {'$ref': 'b'}
    >>> schema_merge(a, b, ('foo',))
    {'type': OrderedSet(['a', 'b'])}

    >>> a, b = {'$ref': 'a'}, {'type': ['b', 'c']}
    >>> schema_merge(a, b, ('foo',))
    {'type': OrderedSet(['a', 'b', 'c'])}

    >>> a, b = {'$ref': 'a'}, {'type': OrderedSet(['b', 'c'])}
    >>> schema_merge(a, b, ('foo',))
    {'type': OrderedSet(['a', 'b', 'c'])}

    >>> a, b = {'type': ['a', 'b']}, {'$ref': 'c'}
    >>> schema_merge(a, b, ('foo',))
    {'type': OrderedSet(['a', 'b', 'c'])}

    >>> a, b = {'type': OrderedSet(['a', 'b'])}, {'$ref': 'c'}
    >>> schema_merge(a, b, ('foo',))
    {'type': OrderedSet(['a', 'b', 'c'])}

    >>> a, b = {'Foo': {'$ref': 'a'}}, {'Foo': {'type': 'b'}}
    >>> schema_merge(a, b, ('foo',))
    {'Foo': {'type': OrderedSet(['a', 'b'])}}

    >>> schema_merge({'type': 'a'}, {'type': 'b'}, ()) # doctest: +NORMALIZE_WHITESPACE
    {'type': OrderedSet(['a', 'b'])}

    >>> schema_merge({'type': 'string'}, {'type': 'integer'}, ())
    {'type': OrderedSet(['string', 'integer'])}
    """
    if not (isinstance(target, Mapping) and isinstance(src, Mapping)):
        raise TypeError("Both schemas must be dictionaries")

    for key, src_schema in src.items():
        try:
            if key in (
                REF,
                TYPE,
            ):  # $ref and type are treated similarly and unified
                target_schema = target.get(key) or target.get(TYPE) or target[REF]
            else:
                target_schema = target[key]  # carry over existing properties
        except KeyError:
            target[key] = src_schema
        else:
            next_path = path + (key,)
            try:
                target[key] = schema_merge(target_schema, src_schema, next_path)
            except TypeError:
                if key in (TYPE, REF):  # combining multiple $ref and types
                    src_set = to_set(src_schema)

                    try:
                        target[TYPE] = to_set(
                            target[TYPE]
                        )  # casting to ordered set as lib
                        # implicitly converts strings to sets
                        target[TYPE] |= src_set
                    except (TypeError, KeyError):
                        target_set = to_set(target_schema)
                        target[TYPE] = target_set | src_set

                    try:
                        # check if there are conflicting $ref and type
                        # at the same sub schema. Conflicting $ref could only
                        # happen on combiners because method merges two json
                        # objects without losing any previous info:
                        # e.g. "oneOf": [{"$ref": "..#1.."},{"$ref": "..#2.."}] ->
                        # { "ref": "..#1..", "type": [{},{}] }
                        target.pop(REF)
                    except KeyError:
                        pass

                elif key == "required":
                    target[key] = sorted(set(target_schema) | set(src_schema))
                else:
                    if key in NON_MERGABLE_KEYS and target_schema != src_schema:
                        msg = (
                            "Object at path '{path}' declared multiple values "
                            "for '{}': found '{}' and '{}'"
                        )
                        # pylint: disable=W0707
                        raise ConstraintError(msg, path, key, target_schema, src_schema)
                    target[key] = src_schema
    return target