Skip to content

symfonic.capabilities.memory.graph_claimed_lifecycle

graph_claimed_lifecycle

Private token-CAS graph-only PG lifecycle, with old adapters unchanged.

discard_pending async 

discard_pending(store: Any, pending: list[Any]) -> list[str]

Discard every unsealed pending row in the caller's lifecycle transaction.

Source code in src/symfonic/capabilities/memory/graph_claimed_lifecycle.py 
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
async def discard_pending(store: Any, pending: list[Any]) -> list[str]:
    """Discard every unsealed pending row in the caller's lifecycle transaction."""
    from symfonic.memory.layers.procedural.lifecycle import procedure_is_sealed_terminal

    kept = [
        node for node in pending
        if not (node.layer.value == "procedural"
                and procedure_is_sealed_terminal(node.properties))
    ]
    owned = _batch_owners(store, kept)
    if owned:
        from symfonic.memory.backends.postgres_claimed_discard import discard_claimed
        from symfonic.memory.backends.postgres_pending_claims import token_of

        await discard_claimed(
            store._graph.pool, owned[0][0].tenant_id,
            [(owner, str(node.id), token_of(node.properties)) for owner, node in owned],
            store._graph._edge_upsert_mode,
        )
        return [str(node.id) for node in kept]
    for node in kept:
        if store._recall is not None:
            await store._recall.forget(_adapter_scope(node_scope(node)), (str(node.id),))
        await discard(store, node)
    return [str(node.id) for node in kept]

publish_pending async 

publish_pending(store: Any, pending: list[Any], batched: bool) -> None

Publish every pending row in the caller's lifecycle transaction.

Source code in src/symfonic/capabilities/memory/graph_claimed_lifecycle.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
async def publish_pending(store: Any, pending: list[Any], batched: bool) -> None:
    """Publish every pending row in the caller's lifecycle transaction."""
    owned = _batch_owners(store, pending)
    if owned:
        from symfonic.memory.backends.postgres_claimed_publish import publish_claimed

        await publish_claimed(store._graph.pool, owned[0][0].tenant_id, owned)
        return
    from symfonic.capabilities.memory import graph_receipt
    from symfonic.capabilities.memory.compat import record_from_legacy_node
    from symfonic.capabilities.memory.graph_rows import node_payload

    for node in pending:
        if store._recall is not None and not batched:
            record = record_from_legacy_node(node_payload(node))
            await graph_receipt.remember(store, _adapter_scope(node_scope(node)), record)
        await publish(store, node)