Python reactions

A reaction receives the diffs a continuous query produces. The simplest is a callable:

def on_results(event):
    for diff in event["results"]:
        if diff["type"] == "ADD":
            print("entered:", diff["data"])
        elif diff["type"] == "DELETE":
            print("left:", diff["data"])
        else:
            print("changed:", diff["before"], "->", diff["after"])


await drasi.add_python_reaction("watch", ["open-orders"], on_results)

One reaction can watch several queries. The event carries which query produced it, so a single handler can serve them all.

Async callbacks

An async callable works the same way, and is awaited on your event loop, so it can do async work of its own:

async def on_results(event):
    async with httpx.AsyncClient() as client:
        for diff in event["results"]:
            await client.post("https://example.com/hook", json=diff)


await drasi.add_python_reaction("webhook", ["open-orders"], on_results)

Which keys a diff carries

type is always present. What comes with it depends on the type:

Reading data on an UPDATE gets you nothing, so branch on type first.

Durable reactions

An ordinary reaction starts from scratch after a restart: whatever was delivered while the process was down is gone. A durable reaction records how far it got, and resumes from there.

It needs two stores, configured when the engine is created. The state store keeps the checkpoint. The index store keeps the query’s outbox, which is what the checkpoint points at. An in-memory query is rejected: after a restart there would be nothing to resume from.

drasi = await Drasi.create(
    "app",
    state_store={"kind": "redb", "path": "/var/lib/myapp/drasi/state.redb"},
    index_store={"kind": "rocksdb", "path": "/var/lib/myapp/drasi/index"},
)

Then register the reaction with a recovery policy:

async def on_results(event): ...


await drasi.add_durable_python_reaction(
    "billing",
    ["open-orders"],
    on_results,
    recovery_policy="strict",
)

The policy decides what happens when the recorded checkpoint cannot be satisfied — for example when the data it referred to is no longer available after a restart:

strict is the default: the alternatives accept data loss.

Checkpoints advance only after your callback returns without raising. If it raises under strict or auto_reset, the reaction stops in an error state and the checkpoint stays put, so the event is delivered again the next time the reaction starts. skip_gap records the event as handled and continues.

Reading results directly

A reaction is push-based. To pull the current result set instead:

rows = await drasi.get_query_results("open-orders")

That returns the full result set as it stands, which is useful for rendering an initial view before switching to diffs.

Removing a reaction

await drasi.remove_reaction("watch")

Removing a reaction does not affect the query. Removing a query that a reaction still depends on is refused, so you cannot leave a reaction subscribed to something that no longer exists.

Next

• Streaming results — the same data as an async iterator.
• Error handling — what a failing callback does.