Skip to content

symfonic.memory.backends.in_memory_bulk_apply

in_memory_bulk_apply

Private InMemory bulk dedup commit; public bulk stays gated.

apply 

apply(store: Any, scope: StorageScope, plan: BulkDedupPlan) -> int

Validate and mutate without yielding; restore both tables on any failure.

This operation joins an active transaction on its owner task. Otherwise its shared lock covers backend graph writers only, so it is not yet a public full-pass contract.

Source code in src/symfonic/memory/backends/in_memory_bulk_apply.py 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
def apply(store: Any, scope: StorageScope, plan: BulkDedupPlan) -> int:
    """Validate and mutate without yielding; restore both tables on any failure.

    This operation joins an active transaction on its owner task. Otherwise its
    shared lock covers backend graph writers only, so it is not yet a public
    full-pass contract.
    """
    from symfonic.core.scope import TenantScope
    from symfonic.memory.janitor_bulk_plan import BulkDedupPlan
    if not isinstance(scope, TenantScope) or not isinstance(plan, BulkDedupPlan):
        raise TypeError("Legacy TenantScope and BulkDedupPlan required")
    _validate(store, scope, plan)
    tenant = scope.tenant_id
    original_nodes = dict(store._nodes.get(tenant, {}))
    saved_nodes = deepcopy(original_nodes)
    original_edges = dict(store._edges.get(tenant, {}))
    try:
        for action in plan.actions:
            if action.kind == "update":
                _update(store, scope, action)
            else:
                _delete(store, scope, action.node_id)
    except BaseException:
        try:
            for identifier, original in original_nodes.items():
                saved = saved_nodes[identifier]
                # saved is already detached; copying again can fail mid-restore.
                fields = dict(saved.__dict__)
                fields_set = saved.__pydantic_fields_set__.copy()
                object.__setattr__(original, "__dict__", fields)
                object.__setattr__(original, "__pydantic_fields_set__", fields_set)
                object.__setattr__(original, "__pydantic_private__",
                                   saved.__pydantic_private__.copy())
        finally:
            store._nodes[tenant] = original_nodes
            store._edges[tenant] = original_edges
        raise
    return plan.merged_count