Connect storage through your callables

Prerequisite: persistence. Melder calls your storage functions; the example uses an in-memory dictionary so the complete integration can run without a database service or credentials.

Handler	Arguments	Responsibility
Store	kind, profile name, unit ID, payload	Store a plain record
Fetch	kind, unit ID	Return the stored payload or a known absence
List	kind, profile name	Enumerate unit IDs for the requested partition
Delete	kind, unit ID	Remove the explicitly identified stored unit

Configure the needed handlers on ExternalPersistenceManagerConfiguration, then attach it through Crystallizer. Inspect describe_external_persistence_manager() for the installed wiring and failure counts; describe_external_interface() describes the integration contract.

Verify the remote leg separately

A successful local flush does not establish remote delivery under the default lenient write policy. The failure lesson deliberately raises in a store handler, checks the local cache, and checks the increased failure count. Missing read lanes refuse rather than report an empty remote record.

The JSON lesson also shows the serialized payload crossing a callable boundary. For record round trips, use strict JSON serialization so unsupported values fail instead of being silently converted into lossy strings.

The workflow in code

These are the core steps from the saved example. Use its complete linked script for all class definitions and setup.

def main() -> None:
    crystallizer = md.Crystallizer()
    crystallizer.activate(
        md.CrystallizerConfigurationBuilder().with_defaults().activate(),
    )

    # THE HANDLERS GO IN THEIR OWN OBJECT, never into the record.
    mesh = md.ExternalPersistenceManagerConfiguration()
    mesh.with_store_handler(store)
    mesh.with_fetch_handler(fetch)
    mesh.with_list_units_handler(list_units)
    crystallizer.configure_external_persistence_manager(mesh)
    print("mesh attached - melder now has four callables and no idea what")
    print("is behind them")

    # WHAT IS WIRED vs WHAT THE CONTRACT IS. Two questions, two doors,
    # neither one a network call.
    wiring = crystallizer.describe_external_persistence_manager()
    contract = crystallizer.describe_external_interface()
    assert isinstance(wiring, dict) and isinstance(contract, dict)
    assert wiring.get("attached") is True, wiring
    print()
    print("describe_external_persistence_manager ->", len(wiring),
          "keys (what is WIRED)")
    print("describe_external_interface           ->", len(contract),
          "keys (what the CONTRACT is)")

    # PRESENCE FLAGS, NEVER CALLABLES. Search the whole record payload for
    # anything callable - a recorded world has to stay code-free.
    leaked = [key for key, value in wiring.items() if callable(value)]
    assert not leaked, "the record leaked a callable: %s" % leaked
    print("  the wiring record holds no callable objects - only presence")
    print("  flags, because executable code cannot be serialized into a")
    print("  world record and a record that embedded yours would only")
    print("  restore where that code already ran")

    # A WORLD WORTH SEALING.
    spellbook_configuration = (
        md.SpellbookConfiguration(FRAME).with_defaults().finalize()
    )
    book = md.Spellbook(aetheric_frame=FRAME,
                        configuration=spellbook_configuration)
    book.configure_aether_frame(
        system_state="dynamic",
        disposal=None,
        disposal_method_names=None,
        rift_enabled=True,
        ai_native=True,
    )
    book.bind(spell=Ledger, existence="unique", permissions="create",
              binding_name="mesh-ledger")
    book.conjure(name="mesh-root")

    # SEAL, THEN SHIP - one verb, two guarantees, one of them yours.
    checkpoint_id = crystallizer.create_checkpoint()
    assert checkpoint_id in crystallizer.list_checkpoint_ids()
    flushed = crystallizer.flush_checkpoint(checkpoint_id)

    # THE LOCAL SEAL IS THE PART THE RETURN VALUE COVERS. Assert it.
    assert checkpoint_id in flushed, flushed
    assert checkpoint_id in crystallizer.list_cached_checkpoint_ids()
    print()
    print("flush_checkpoint ->", checkpoint_id[:14], "...")
    print("  LOCAL SEAL asserted: it is in the cache, on this machine")

    # THE REMOTE LEG IS LENIENT, so this is REPORTED and not asserted.
    # Teaching you to assert it would be teaching you to trust the half
    # the return value does not cover.
    print("  your database received", len(WRITES), "unit(s):")
    for kind, profile_name, unit_id in WRITES[:4]:
        print("     kind=%-11s profile=%-10s unit=%s"
              % (kind, profile_name, unit_id[:14]))
    print("  reported, NOT asserted - the remote leg is lenient by")
    print("  default, so a clean return proves the local seal and nothing")
    print("  about your storage. Local custody is never hostage to a")
    print("  network you do not own. If you need remote confirmation, the")
    print("  return value of flush is not it - strict uploads are")

    # AND THE ABSENCE THAT MATTERS. No sync(), no mirror_all().
    for absent in ("sync", "mirror", "mirror_all", "sync_external",
                   "push_everything"):
        assert not hasattr(crystallizer, absent), (
            "%s appeared - an opaque sync verb would make the mesh "
            "impossible to reason about when it disagrees with itself"
            % absent
        )
    print()
    print("there is NO sync() - every verb names a KIND and a DIRECTION,")
    print("so when the two sides disagree you can say which way the last")
    print("byte was travelling. `sync` cannot answer that question.")

    print()
    print("four callables, one kind column, zero database imports")

Runnable examples

Getting data into your database — Expert 09
Persistence mesh your code their lanes — Expert 02
The record crosses as a json string — Expert 28

All expert examples · Level contents · Full contents

Canonical page source