API Reference¶

Merge¶

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

Parameters:

schema (str | Dict[str, Any] | None)
merge_rules (Dict[Tuple[str, ...], str] | None)
rule_overrides (Dict[Tuple[str, ...], MergeStrategy] | None)

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

Initializes 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]¶

Merges a list of releases into a compiled release.

Parameters:: releases (List[Dict[str, Any]])
Return type:: Dict[str, Any]

create_versioned_release(releases)[source]¶

Merges a list of releases into a versioned release.

Parameters:: releases (List[Dict[str, Any]])
Return type:: Dict[str, Any]

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 (str | Dict[str, Any] | None)
merge_rules (Dict[Tuple[str, ...], str] | None)
rule_overrides (Dict[Tuple[str, ...], MergeStrategy] | None)

versioned: bool | None = None¶

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

Initializes 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]¶

Returns the merged release as a dictionary.

Return type:: Dict[str, Any]

extend(releases)[source]¶

Sorts and merges many releases into the merged release.

Parameters:: releases (List[Dict[str, Any]])
Return type:: None

append(release)[source]¶

Merges 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': typing.Optional[bool]}¶

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

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

versioned: bool | None = False¶

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

Initializes 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

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

Parameters:

data (Dict[str, Any] | None)
schema (str | Dict[str, Any] | None)
merge_rules (Dict[Tuple[str, ...], str] | None)
rule_overrides (Dict[Tuple[str, ...], MergeStrategy] | None)

versioned: bool | None = True¶

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]¶

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”).

Parameters:: schema (str | Dict[str, Any] | None)
Return type:: Dict[Tuple[str, …], str]

Flatten¶

class ocdsmerge.flatten.MergeStrategy(value)[source]¶

An enumeration.

APPEND = 1¶

MERGE_BY_POSITION = 2¶

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

A string with identifier and original_value properties.

Parameters:: identifier (int | str)

property original_value: int | str | None¶

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

Returns 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]¶

Flattens a JSON object into key-value pairs, in which the key is the JSON path as a tuple. For example:

Replaces numbers in JSON paths (representing positions in arrays) with special objects. This ensures that 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 (Dict[Tuple[str, ...], str])
rule_overrides (Dict[Tuple[str, ...], MergeStrategy])
flattened (Dict[Tuple[int | str, ...], Any])
path (Tuple[int | str, ...])
rule_path (Tuple[str, ...])
versioned (bool | None)

Return type:

Dict[Tuple[int | str, …], Any]

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]¶

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

Return type:: List[str]

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

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

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

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

Sorts 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 and an object 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