Skip to content

symfonic.memory.backends.in_memory_node_update

in_memory_node_update

Existing unconditional reference node updates, extracted for module budget.

update 

update(store: Any, scope: Any, node_id: Any, updates: dict[str, Any]) -> Any

Caller holds the existing backend writer lock.

Source code in src/symfonic/memory/backends/in_memory_node_update.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
def update(store: Any, scope: Any, node_id: Any, updates: dict[str, Any]) -> Any:
    """Caller holds the existing backend writer lock."""
    node = store._nodes.get(scope.tenant_id, {}).get(str(node_id))
    if node is None or not store._node_visible(scope, node):
        raise KeyError(f"Node {node_id} not found for tenant {scope.tenant_id}")
    safe = in_memory_updates(deepcopy(updates), node.properties.get(SCOPE_PATH_KEY))
    if any(key in type(node).__private_attributes__ for key in safe):
        raise ValueError("private node fields cannot be updated")
    guard_update(node, safe)
    for key, value in safe.items():
        if key == "updated_at" and isinstance(value, str):
            value = datetime.fromisoformat(value)
        if key in type(node).model_fields:
            object.__setattr__(node, key, value)
        else:
            node.properties[key] = value
    if "updated_at" not in updates:
        node.updated_at = datetime.now(UTC)
    from symfonic.memory.backends.in_memory_owned_node_patch import advance
    advance(node)
    return node.model_copy(deep=True)