How to Merge Incremental Snapshot Upsert Dict in Python

Merge a snapshot dict into a base dict, recursively updating nested dictionaries while preferring snapshot values on conflicts.

Easy Python 3.9+ Aug 9, 2026 Data pipelines & processing 13 views 0 copies

dict merge upsert recursion data-pipeline

Python code

34 lines

Python 3.9+

def merge_upsert(base: dict, snapshot: dict) -> dict:
    """
    Merge a snapshot dict into a base dict, preferring snapshot values 
    on key conflicts (upsert semantics). Nested dicts are merged recursively.
    """
    result = dict(base)
    
    for key, value in snapshot.items():
        if key in result and isinstance(result[key], dict) and isinstance(value, dict):
            # Recursively merge nested dicts
            result[key] = merge_upsert(result[key], value)
        else:
            # Upsert overwrites
            result[key] = value
    
    return result


if __name__ == "__main__":
    base = {
        "id": 1,
        "name": "Alice",
        "profile": {"age": 30, "city": "NYC"},
        "tags": ["old"]
    }
    
    snapshot = {
        "name": "Alice Smith",
        "profile": {"city": "Boston"},
        "active": True
    }
    
    merged = merge_upsert(base, snapshot)
    print(merged)

Output

stdout

{'id': 1, 'name': 'Alice Smith', 'profile': {'age': 30, 'city': 'Boston'}, 'tags': ['old'], 'active': True}

How it works

The function merge_upsert starts by copying the base dict to avoid mutating the input. It iterates through each key-value pair in the snapshot. If the key already exists in the result and both the existing value and the snapshot value are dictionaries, it recursively merges them, preserving nested fields from the base that are not in the snapshot. Otherwise, it overwrites the base value with the snapshot value, effectively upserting the new data. This pattern is common in data pipelines where a base state needs to be updated with incremental changes.

Common mistakes

Forgetting to copy the base dict, which mutates the original input.
Assuming all values are dicts when checking for nested merges, causing type errors.
Overwriting nested dicts entirely instead of merging them recursively.

Variations

Use a loop with `update()` for a shallow merge when nested dicts are not needed.
Implement with `collections.ChainMap` for a read-only view that prefers snapshot.

Real-world use cases

Applying daily data updates to a master record without losing fields not in the update.
Merging configuration changes from a service deployment into existing runtime settings.
Combining partial user profile updates from an API into internal stored state.

How to Merge Incremental Snapshot Upsert Dict in Python

Python code

Output

How it works

Common mistakes

Variations

Real-world use cases

More from Data pipelines & processing

Tutorials

Quizzes

Python code

Output

How it works

Common mistakes

Variations

Real-world use cases

More from Data pipelines & processing

Keep learning

Tutorials

Quizzes