Skip to content

Diff Utilities

The pydoover utilities module provides functions for generating and applying diffs between dictionary objects. These are particularly useful for synchronizing state, tracking changes, and transmitting only modified data over the network.

Import

from pydoover.utils import (
    apply_diff,
    generate_diff,
    maybe_load_json,
)

Overview

The diff utilities implement a recursive diff format compatible with the Doover platform:

  • Diffs contain only changed keys - Unchanged values are not included
  • Nested objects are recursively diffed - Changes at any depth are captured
  • Deleted keys are represented as None - Setting a key to None in the diff removes it from the target

This format is efficient for:

  • Synchronizing state between devices and cloud
  • Tracking configuration changes
  • Minimizing data transmission

apply_diff()

Applies a diff to a dictionary, returning the updated result.

Signature

def apply_diff(
    data: dict[str, Any],
    diff: dict[str, Any],
    do_delete: bool = True,
    clone: bool = True,
) -> dict[str, Any]

Parameters

ParameterTypeDefaultDescription
datadictrequiredThe original dictionary to update
diffdictrequiredThe diff to apply
do_deleteboolTrueIf True, None values in diff delete keys; if False, sets them to None
cloneboolTrueIf True, creates a deep copy; if False, modifies in-place

Return Value

Returns a new dictionary with the diff applied (or the modified original if clone=False).

Examples

Basic Update

from pydoover.utils import apply_diff

data = {
    "name": "sensor-01",
    "temperature": 25.5,
    "status": "active"
}

diff = {
    "temperature": 26.0,
    "status": "warning"
}

result = apply_diff(data, diff)
print(result)
# {
#     "name": "sensor-01",
#     "temperature": 26.0,
#     "status": "warning"
# }

Adding New Keys

data = {"a": 1}
diff = {"b": 2, "c": 3}

result = apply_diff(data, diff)
print(result)
# {"a": 1, "b": 2, "c": 3}

Deleting Keys

data = {"a": 1, "b": 2, "c": 3}
diff = {"b": None}  # Delete key "b"

result = apply_diff(data, diff)
print(result)
# {"a": 1, "c": 3}

Nested Updates

data = {
    "sensors": {
        "temp": {"value": 25, "unit": "C"},
        "humidity": {"value": 60, "unit": "%"}
    },
    "meta": {"version": 1}
}

diff = {
    "sensors": {
        "temp": {"value": 26}  # Only update temp value
    }
}

result = apply_diff(data, diff)
print(result)
# {
#     "sensors": {
#         "temp": {"value": 26, "unit": "C"},  # value updated, unit preserved
#         "humidity": {"value": 60, "unit": "%"}  # unchanged
#     },
#     "meta": {"version": 1}  # unchanged
# }

In-Place Modification

data = {"a": 1, "b": 2}
diff = {"b": 3}

# Modify data directly (no copy)
result = apply_diff(data, diff, clone=False)

# data and result are the same object
print(data is result)  # True
print(data)  # {"a": 1, "b": 3}

Preserving None Values

data = {"a": 1, "b": 2}
diff = {"b": None}

# With do_delete=False, None is set as the value
result = apply_diff(data, diff, do_delete=False)
print(result)
# {"a": 1, "b": None}

generate_diff()

Generates a diff between two dictionaries.

Signature

def generate_diff(
    old: dict[str, Any],
    new: dict[str, Any],
    do_delete: bool = True,
) -> dict[str, Any]

Parameters

ParameterTypeDefaultDescription
olddictrequiredThe original dictionary
newdictrequiredThe new dictionary
do_deleteboolTrueIf True, deleted keys are marked as None

Return Value

Returns a dictionary containing only the differences. Keys that are the same in both dictionaries are not included.

Examples

Basic Diff

from pydoover.utils import generate_diff

old = {"a": 1, "b": 2, "c": 3}
new = {"a": 1, "b": 5, "c": 3}

diff = generate_diff(old, new)
print(diff)
# {"b": 5}  # Only changed key

Detecting Additions

old = {"a": 1}
new = {"a": 1, "b": 2, "c": 3}

diff = generate_diff(old, new)
print(diff)
# {"b": 2, "c": 3}

Detecting Deletions

old = {"a": 1, "b": 2, "c": 3}
new = {"a": 1}

diff = generate_diff(old, new)
print(diff)
# {"b": None, "c": None}  # Deleted keys marked as None

Nested Diff

old = {
    "config": {
        "timeout": 30,
        "retries": 3,
        "debug": False
    }
}

new = {
    "config": {
        "timeout": 60,  # Changed
        "retries": 3,   # Same
        "debug": True   # Changed
    }
}

diff = generate_diff(old, new)
print(diff)
# {
#     "config": {
#         "timeout": 60,
#         "debug": True
#     }
# }

No Changes

old = {"a": 1, "b": 2}
new = {"a": 1, "b": 2}

diff = generate_diff(old, new)
print(diff)
# {}  # Empty diff means no changes

Disable Delete Markers

old = {"a": 1, "b": 2}
new = {"a": 1}

# Without delete markers
diff = generate_diff(old, new, do_delete=False)
print(diff)
# {}  # "b" deletion not recorded

maybe_load_json()

Attempts to parse a string as JSON. Returns the original value if parsing fails.

Signature

def maybe_load_json(data: Any) -> Any

Parameters

ParameterTypeDescription
dataAnyValue to attempt JSON parsing on

Return Value

Returns the parsed JSON object if successful, otherwise returns the original value.

Examples

from pydoover.utils import maybe_load_json

# Valid JSON string
result = maybe_load_json('{"name": "test", "value": 42}')
print(result)
# {"name": "test", "value": 42}

# Invalid JSON - returns original
result = maybe_load_json("not json")
print(result)
# "not json"

# Already a dict - returns as-is
result = maybe_load_json({"already": "dict"})
print(result)
# {"already": "dict"}

Common Patterns

State Synchronization

from pydoover.utils import apply_diff, generate_diff

class StateManager:
    def __init__(self):
        self.state = {}
        self.last_synced_state = {}

    def update_local(self, updates: dict):
        """Apply updates to local state"""
        self.state = apply_diff(self.state, updates)

    def get_pending_changes(self) -> dict:
        """Get changes since last sync"""
        return generate_diff(self.last_synced_state, self.state)

    def apply_remote_changes(self, diff: dict):
        """Apply changes from remote"""
        self.state = apply_diff(self.state, diff)
        self.last_synced_state = apply_diff(self.last_synced_state, diff)

    def mark_synced(self):
        """Mark current state as synced"""
        self.last_synced_state = apply_diff({}, self.state)  # Deep copy

Configuration Updates

from pydoover.utils import apply_diff, generate_diff

class ConfigManager:
    def __init__(self, default_config: dict):
        self.config = default_config.copy()

    def update(self, changes: dict):
        """Update configuration with changes"""
        old_config = apply_diff({}, self.config)  # Copy
        self.config = apply_diff(self.config, changes)

        # Log what changed
        diff = generate_diff(old_config, self.config)
        if diff:
            print(f"Configuration updated: {diff}")

    def reset_to_defaults(self, defaults: dict):
        """Reset to defaults, preserving non-default keys"""
        self.config = apply_diff(defaults, self.config)

Efficient Cloud Updates

from pydoover.docker import Application
from pydoover.utils import generate_diff, apply_diff

class EfficientApp(Application):
    def setup(self):
        self.device_state = {}
        self.last_published_state = {}

    async def publish_state_changes(self):
        """Only publish what has changed"""
        diff = generate_diff(self.last_published_state, self.device_state)

        if diff:  # Only publish if there are changes
            await self.device_agent.publish(
                "device_state",
                diff
            )
            self.last_published_state = apply_diff(
                self.last_published_state,
                diff
            )

    async def handle_remote_update(self, diff: dict):
        """Apply remote updates to local state"""
        self.device_state = apply_diff(self.device_state, diff)

Change Detection

from pydoover.utils import generate_diff

def has_changes(old: dict, new: dict) -> bool:
    """Check if there are any differences"""
    diff = generate_diff(old, new)
    return bool(diff)

def get_changed_keys(old: dict, new: dict) -> set:
    """Get the set of keys that changed"""
    diff = generate_diff(old, new)
    return set(diff.keys())

Merging Updates

from pydoover.utils import apply_diff

def merge_updates(*updates: dict) -> dict:
    """Merge multiple update dicts into one"""
    result = {}
    for update in updates:
        result = apply_diff(result, update)
    return result

# Usage
update1 = {"a": 1, "b": 2}
update2 = {"b": 3, "c": 4}
update3 = {"c": 5, "d": 6}

merged = merge_updates(update1, update2, update3)
print(merged)
# {"a": 1, "b": 3, "c": 5, "d": 6}

Edge Cases

Non-Dict Values

When either operand is not a dict, the behavior changes:

# If diff is not a dict, it replaces data entirely
result = apply_diff({"a": 1}, "new value")
print(result)  # "new value"

# If data is not a dict, diff is returned
result = apply_diff("old value", {"a": 1})
print(result)  # {"a": 1}

Empty Dicts

# Applying empty diff returns unchanged data
result = apply_diff({"a": 1}, {})
print(result)  # {"a": 1}

# Diff of identical dicts is empty
diff = generate_diff({"a": 1}, {"a": 1})
print(diff)  # {}

Deeply Nested Structures

data = {
    "level1": {
        "level2": {
            "level3": {
                "value": 100
            }
        }
    }
}

diff = {
    "level1": {
        "level2": {
            "level3": {
                "value": 200
            }
        }
    }
}

result = apply_diff(data, diff)
# Only the deeply nested value is changed

Best Practices

  1. Use clone=True for safety - Unless you specifically need in-place modification
  2. Check for empty diffs - Avoid unnecessary operations when nothing changed
  3. Be consistent with do_delete - Use the same setting for generate_diff and apply_diff
  4. Handle None values carefully - Remember that None means "delete" by default
  5. Consider JSON serialization - The diff format is JSON-compatible for network transmission

Related Pages