Source code for ocdsmerge.rules

from functools import lru_cache
from typing import Any, Dict, Generator, List, Optional, Tuple, Union

import jsonref

from ocdsmerge.util import get_release_schema_url, get_tags

MergeRules = Dict[Tuple[str, ...], str]
Schema = Optional[Union[str, Dict[str, Any]]]




def get_merge_rules(schema: Schema = None) -> MergeRules:
    """
    Returns merge rules as key-value pairs, in which the key is a JSON path as a tuple, and the value is the merge rule
    as a string ("omitWhenMerged" or "wholeListMerge").
    """
    schema = schema or get_release_schema_url(get_tags()[-1])
    if isinstance(schema, dict):
        # jsonref.JsonRef is deprecated, but used for backwards-compatibility with jsonref 0.x.
        return _get_merge_rules_from_dereferenced_schema(jsonref.JsonRef.replace_refs(schema))
    else:
        return _get_merge_rules_from_url_or_path(schema)



@lru_cache
def _get_merge_rules_from_url_or_path(schema: str) -> MergeRules:
    if schema.startswith('http'):
        deref_schema = jsonref.load_uri(schema)
    else:
        with open(schema) as f:
            deref_schema = jsonref.load(f)
    return _get_merge_rules_from_dereferenced_schema(deref_schema)


def _get_merge_rules_from_dereferenced_schema(deref_schema: Dict[str, Any]) -> MergeRules:
    return dict(_get_merge_rules(deref_schema['properties']))


def _get_merge_rules(
    properties: Dict[str, Any], path: Optional[Tuple[str, ...]] = None
) -> Generator[Tuple[Tuple[str, ...], str], None, None]:
    """
    Yields merge rules as key-value pairs, in which the first element is a JSON path as a tuple, and the second element
    is the merge rule as a string ("omitWhenMerged" or "wholeListMerge").
    """
    if path is None:
        path = ()

    for key, value in properties.items():
        new_path = path + (key,)
        types = _get_types(value)

        # `omitWhenMerged` supersedes all other rules.
        # See https://standard.open-contracting.org/1.1/en/schema/merging/#discarded-fields
        if value.get('omitWhenMerged') or value.get('mergeStrategy') == 'ocdsOmit':
            yield new_path, 'omitWhenMerged'
        # `wholeListMerge` supersedes any nested rules.
        # See https://standard.open-contracting.org/1.1/en/schema/merging/#whole-list-merge
        elif 'array' in types and (value.get('wholeListMerge') or value.get('mergeStrategy') == 'ocdsVersion'):
            yield new_path, 'wholeListMerge'
        # See https://standard.open-contracting.org/1.1/en/schema/merging/#object-values
        elif 'object' in types and 'properties' in value:
            yield from _get_merge_rules(value['properties'], path=new_path)
        # See https://standard.open-contracting.org/1.1/en/schema/merging/#whole-list-merge
        elif 'array' in types and 'items' in value:
            item_types = _get_types(value['items'])
            if any(item_type != 'object' for item_type in item_types):
                yield new_path, 'wholeListMerge'
            elif 'object' in item_types and 'properties' in value['items']:
                if 'id' not in value['items']['properties']:
                    yield new_path, 'wholeListMerge'
                else:
                    yield from _get_merge_rules(value['items']['properties'], path=new_path)


def _get_types(prop: Dict[str, Any]) -> List[str]:
    """
    Returns a property's `type` as a list.
    """
    if 'type' not in prop:
        return []
    if isinstance(prop['type'], str):
        return [prop['type']]
    return prop['type']