API Reference¶

Merge¶

class ocdsmerge.merge.Merger(schema=None, merge_rules=None, rule_overrides=None)[source]¶

Parameters:

schema (Schema)
merge_rules (MergeRules | None)
rule_overrides (RuleOverrides | None)

__init__(schema=None, merge_rules=None, rule_overrides=None)[source]¶

Initialize a reusable Merger instance for creating merged releases.

Parameters:

schema (dict or str) – the release schema (if not provided, will default to the latest version of OCDS)
merge_rules (dict[tuple[str, ...], str] | None) – the merge rules (if not provided, will determine the rules from the schema)
rule_overrides (dict[tuple[str, ...], MergeStrategy] | None) – any rule overrides, in which keys are field paths as tuples, and values are either ocdsmerge.APPEND or ocdsmerge.MERGE_BY_POSITION

create_compiled_release(releases)[source]¶

Merge a list of releases into a compiled release.

Parameters:: releases (list[dict[str, Any]])
Return type:: dict[str, Any]

create_versioned_release(releases)[source]¶

Merge a list of releases into a versioned release.

Parameters:: releases (list[dict[str, Any]])
Return type:: dict[str, Any]

__annotate_func__ = None¶

__annotations_cache__ = {}¶

__firstlineno__ = 10¶

__static_attributes__ = ('merge_rules', 'rule_overrides')¶

class ocdsmerge.merge.MergedRelease(data=None, schema=None, merge_rules=None, rule_overrides=None)[source]¶

Whether the class is for merging versioned releases.

Parameters:

data (dict[str, Any] | None)
schema (Schema)
merge_rules (MergeRules | None)
rule_overrides (RuleOverrides | None)

versioned: bool | None = None¶

__init__(data=None, schema=None, merge_rules=None, rule_overrides=None)[source]¶

Initialize a merged release.

Parameters:

data (dict[str, Any] | None) – the latest copy of the merged release, if any
schema (dict or str) – the release schema (if not provided, will default to the latest version of OCDS)
merge_rules (dict[tuple[str, ...], str] | None) – the merge rules (if not provided, will determine the rules from the schema)
rule_overrides (dict[tuple[str, ...], MergeStrategy] | None) – any rule overrides, in which keys are field paths as tuples, and values are either ocdsmerge.APPEND or ocdsmerge.MERGE_BY_POSITION

asdict()[source]¶

Return the merged release as a dictionary.

Return type:: dict[str, Any]

extend(releases)[source]¶

Sort and merge many releases into the merged release.

Parameters:: releases (list[dict[str, Any]])
Return type:: None

append(release)[source]¶

Merge one release into the merged release.

Parameters:: release (dict[str, Any])
Return type:: None

flat_append(flat, ocid, release_id, date, tag)[source]¶

Parameters:

flat (dict[tuple[int | str, ...], Any])
ocid (str | None)
release_id (str | None)
date (str | None)
tag (str | None)

Return type:

None

__annotations__ = {'versioned': 'bool | None'}¶

__firstlineno__ = 48¶

__static_attributes__ = ('data', 'merge_rules', 'rule_overrides')¶

class ocdsmerge.merge.CompiledRelease(data=None, **kwargs)[source]¶

Parameters:: data (dict[str, Any] | None)

versioned: bool | None = False¶

__init__(data=None, **kwargs)[source]¶

Initialize a merged release.

Parameters:

data (dict[str, Any] | None) – the latest copy of the merged release, if any
schema (dict or str) – the release schema (if not provided, will default to the latest version of OCDS)
merge_rules – the merge rules (if not provided, will determine the rules from the schema)
rule_overrides – any rule overrides, in which keys are field paths as tuples, and values are either ocdsmerge.APPEND or ocdsmerge.MERGE_BY_POSITION

flat_append(flat, ocid, release_id, date, tag)[source]¶

Parameters:

flat (dict[tuple[int | str, ...], Any])
ocid (str | None)
release_id (str | None)
date (str | None)
tag (str | None)

Return type:

None

__annotate_func__ = None¶

__annotations_cache__ = {}¶

__firstlineno__ = 117¶

__static_attributes__ = ()¶

class ocdsmerge.merge.VersionedRelease(data=None, schema=None, merge_rules=None, rule_overrides=None)[source]¶

Parameters:

data (dict[str, Any] | None)
schema (Schema)
merge_rules (MergeRules | None)
rule_overrides (RuleOverrides | None)

versioned: bool | None = True¶

__annotate_func__ = None¶

__annotations_cache__ = {}¶

__firstlineno__ = 141¶

__static_attributes__ = ()¶

flat_append(flat, ocid, release_id, date, tag)[source]¶

Parameters:

flat (dict[tuple[int | str, ...], Any])
ocid (str | None)
release_id (str | None)
date (str | None)
tag (str | None)

Return type:

None

Rules¶

ocdsmerge.rules.get_merge_rules(schema=None)[source]¶

Return merge rules as key-value pairs.

The key is a JSON path as a tuple, and the value is the merge rule as a string (“omitWhenMerged” or “wholeListMerge”).

Parameters:: schema (str | dict[str, Any] | None)
Return type:: dict[tuple[str, …], str]

Flatten¶

class ocdsmerge.flatten.MergeStrategy(*values)[source]¶

APPEND = 1¶

MERGE_BY_POSITION = 2¶

class ocdsmerge.flatten.IdValue(identifier)[source]¶

A string with identifier and original_value properties.

Parameters:: identifier (Identifier)

identifier¶

property original_value: int | str | None¶

ocdsmerge.flatten.is_versioned_value(value)[source]¶

Return whether the value is a versioned value.

Parameters:: value (dict[str, Any])
Return type:: bool

ocdsmerge.flatten.flatten(obj, merge_rules, rule_overrides, flattened, path=(), rule_path=(), *, versioned=False)[source]¶

Flatten a JSON object into key-value pairs, in which the key is the JSON path as a tuple.

It replaces numbers in JSON paths (representing positions in arrays) with special objects. This ensures objects in arrays with different id values have different JSON paths - and makes it easy to identify such arrays.

{
    "a": "I am a",
    "b": ["A", "list"],
    "c": [
        {"id": 1, "cb": "I am ca"},
        {"id": 2, "ca": "I am cb"}
    ]
}

flattens to:

{
    ('a',): 'I am a',
    ('b',): ['A', 'list'],
    ('a', '1', 'cb'): 'I am ca',
    ('a', '1', 'id'): 1,
    ('a', '2', 'ca'): 'I am cb',
    ('a', '2', 'id'): 2,
}

Parameters:

obj (list[dict[str, Any]] | dict[str, Any])
merge_rules (MergeRules)
rule_overrides (RuleOverrides)
flattened (Flattened)
path (tuple[Identifier, ...])
rule_path (tuple[str, ...])
versioned (bool | None)

Return type:

Flattened

ocdsmerge.flatten.unflatten(flattened)[source]¶

Unflattens a flattened object into a JSON object.

Parameters:: flattened (dict[tuple[int | str, ...], Any])
Return type:: dict[str, Any]

Utilities¶

ocdsmerge.util.get_tags()[source]¶

Return the tags of all versions of OCDS in alphabetical order.

Return type:: list[str]

ocdsmerge.util.get_release_schema_url(tag)[source]¶

Return the URL of the release schema in the given version of OCDS.

Parameters:: tag (str)
Return type:: str

ocdsmerge.util.sorted_releases(releases)[source]¶

Sort a list of releases by date.

Parameters:: releases (list[dict[str, Any]])
Return type:: list[dict[str, Any]]

Exceptions¶

exception ocdsmerge.exceptions.OCDSMergeError[source]¶: Base class for exceptions from within this package.

exception ocdsmerge.exceptions.MissingDateKeyError(key, message)[source]¶

Raised when a release is missing a ‘date’ key.

Parameters:

key (str)
message (str)

exception ocdsmerge.exceptions.NonObjectReleaseError[source]¶: Raised when a release is not an object.

exception ocdsmerge.exceptions.NullDateValueError[source]¶: Raised when a release has a null ‘date’ value.

exception ocdsmerge.exceptions.NonStringDateValueError[source]¶: Raised when a release has a non-string ‘date’ value.

exception ocdsmerge.exceptions.InconsistentTypeError[source]¶: Raised when a path is a literal, an object, and/or an array in different releases.

exception ocdsmerge.exceptions.OCDSMergeWarning[source]¶: Base class for warnings from within this package.

exception ocdsmerge.exceptions.DuplicateIdValueWarning[source]¶: Used when at least two objects in the same array have the same value for the ‘id’ field.