Skip to content

symfonic.capabilities.memory.receipt_stage_values

receipt_stage_values

Small receipt-stage value guards shared by preparation and publication.

flush_with_source async 

flush_with_source(writer, scope, operation, source)

Keep Mongo/ordinary writers unchanged; PG rechecks source at commit.

Source code in src/symfonic/capabilities/memory/receipt_stage_values.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
async def flush_with_source(writer, scope, operation, source):
    """Keep Mongo/ordinary writers unchanged; PG rechecks source at commit."""
    from symfonic.memory.backends.postgres_receipt_write import PostgresReceiptWriter

    if type(writer) is PostgresReceiptWriter:
        return await writer.flush(scope, operation=operation, write=source)
    if type(source) is MemoryOperation and operation.selected_manifest is not None:
        from symfonic.memory.backends.postgres_receipt_selected import (
            operation_for_selected_write,
        )

        if operation_for_selected_write(
            operation.operation_id, source, operation.selected_manifest,
        ) == operation:
            raise _refusal()
    return await writer.flush(scope, operation=operation)

source_bound_intent 

source_bound_intent(session, source, intent)

Add the exact committed WRITE only for the PostgreSQL Agent journal.

Source code in src/symfonic/capabilities/memory/receipt_stage_values.py 
31
32
33
34
35
36
37
38
39
40
41
42
def source_bound_intent(session, source, intent):
    """Add the exact committed WRITE only for the PostgreSQL Agent journal."""
    if not _native_pg(session._journal):
        return intent
    if (type(source) is not MemoryOperation
            or source.verb is not MemoryOperationVerb.WRITE
            or source.scope != session._owner.full_scope):
        raise _refusal()
    return {**intent, "accepted_write": {
        "operation_id": source.operation_id,
        "intent_digest": source.intent_digest,
    }}