colander_data_converter.base.utils

class colander_data_converter.base.utils.BaseModelMerger(strategy=MergingStrategy.OVERWRITE)[source]

Bases: object

A utility class for merging pydantic.BaseModel instances with configurable strategies.

This class provides functionality to merge fields from a source BaseModel into a destination BaseModel, handling both regular model fields and extra attributes. Fields containing ObjectReference types are automatically excluded from merging and reported as unprocessed.

The merger supports two strategies:

  • PRESERVE: Only merge fields if the destination field is empty or None

  • OVERWRITE: Always merge fields from source to destination

Fields are merged based on type compatibility and field constraints. Extra attributes are automatically converted to strings when stored in the attribute dictionary (if supported by the destination model).

Example

>>> from pydantic import BaseModel
>>> class SourceModel(BaseModel):
...     name: str
...     age: int
...     attributes: dict = {}
>>> class DestinationModel(BaseModel):
...     name: str
...     age: int
...     city: str = "Unknown"
...     attributes: dict = {}
>>> source = SourceModel(name="Alice", age=30)
>>> destination = DestinationModel(name="Bob", age=25)
>>> merger = BaseModelMerger(strategy=MergingStrategy.OVERWRITE)
>>> unprocessed = merger.merge(source, destination)
>>> print(destination.name)
Alice
>>> print(destination.age)
30
>>> print(destination.city)
Unknown

Note

  • Fields with ObjectReference types are never merged and are reported as unprocessed

  • Frozen fields cannot be modified and will be reported as unprocessed

  • Complex types (list, dict, tuple, set) in extra attributes are not supported

  • Extra attributes are converted to strings when stored

__init__(strategy=MergingStrategy.OVERWRITE)[source]

Initialize the BaseModelMerger with a merging strategy.

Parameters:

strategy (MergingStrategy) – The strategy to use when merging fields.

merge(source, destination, ignored_fields=None)[source]

Merge all compatible fields from the source object into the destination object.

This method iterates through all fields in the source object and attempts to merge them into the destination object. It handles both regular object fields and extra attributes dictionary if supported.

Parameters:

  • source (BaseModel) – The source model to merge from

  • destination (BaseModel) – The destination model to merge to

  • ignored_fields (List[str] | None) – List of field names to skip during merging

Returns:

A list of field names that could not be processed during the merge operation. Fields containing ObjectReference types are automatically added to this list.

Return type:

List[str]

merge_field(destination, field_name, field_value, ignored_fields=None)[source]

Merge a single field from source to destination model.

This method handles the logic for merging individual fields, including type checking, field existence validation, and attribute handling. It processes both regular model fields and extra attributes based on the destination model’s capabilities and field constraints.

Note

The method follows these rules:

  • Skips fields listed in ignored_fields

  • Skips empty/None field values

  • For fields not in the destination model schema: stores as string in attributes dict (if supported) unless the value is a complex type

  • For schema fields: merges only if type-compatible, not frozen, not containing ObjectReference, and destination is empty (PRESERVE) or strategy is OVERWRITE

Parameters:

  • destination (BaseModel) – The target model to merge into.

  • field_name (str) – The name of the field to merge.

  • field_value (Any) – The value to merge from the source.

  • ignored_fields (List[str] | None) – List of field names to skip during merging.

Returns:

True if the field was processed (successfully merged or handled), False if the field could not be processed

Return type:

bool

class colander_data_converter.base.utils.FeedMerger(source_feed, destination_feed)[source]

Bases: object

merge(delete_unlinked=False, aggressive=False)[source]

Merge the source feed into the destination feed. This method performs a comprehensive merge operation between two ColanderFeeds, handling entity merging, relation updates, and immutable relation management. The merge process follows these main steps:

  1. Identify merging candidates based on entity similarity

  2. Merge compatible entities or add new ones to the destination feed

  3. Update immutable relation references after merging

  4. Copy non-immutable relations from source to destination

Parameters:

  • delete_unlinked (bool) – If True, delete relations involving missing entities

  • aggressive (bool) – If True, temporarily breaks and rebuilds immutable relations to allow more flexible merging. Use with caution as this may affect data integrity during the merge process.

class colander_data_converter.base.utils.MergingStrategy(*values)[source]

Bases: str, Enum

OVERWRITE = 'overwrite'
PRESERVE = 'preserve'