Insert or increment an edge matching scope, endpoints, and relation.
Source code in src/symfonic/memory/backends/in_memory_owned_edge.py
| def upsert(store: Any, scope: TenantScope, edge: Any) -> Any:
"""Insert or increment an edge matching scope, endpoints, and relation."""
from symfonic.memory.backends.in_memory_ownership import check_edge_endpoints
check_edge_endpoints(store, scope, edge)
tenant_edges = store._edges[scope.tenant_id]
path = materialise_scope_path(scope)
conflict = tenant_edges.get(str(edge.id))
if conflict is not None and getattr(conflict, "_scope_path", None) != path:
raise ValueError("Edge ID already belongs to a different scope")
for existing in tenant_edges.values():
if (
getattr(existing, "_scope_path", None) == path
and str(existing.source) == str(edge.source)
and str(existing.target) == str(edge.target)
and existing.relationship == edge.relationship
):
object.__setattr__(existing, "weight", existing.weight + 1)
return existing
edge.tenant_id = scope.tenant_id
edge._scope_path = path
tenant_edges[str(edge.id)] = edge
return edge
|