Python sources

A Python source is one you feed yourself. Register it, then call push_change when your application changes something worth querying.

await drasi.add_python_source("orders")

Pushing a node

await drasi.push_change(
    "orders",
    {
        "op": "insert",
        "id": "o1",
        "labels": ["Order"],
        "properties": {"id": "o1", "status": "open", "total": 42},
    },
)

op is one of insert, update or delete. The aliases add and remove are accepted, and the value is case-insensitive, so INSERT works too.

Updating and deleting

An update replaces the node’s properties:

await drasi.push_change(
    "orders",
    {
        "op": "update",
        "id": "o1",
        "labels": ["Order"],
        "properties": {"id": "o1", "status": "shipped", "total": 42},
    },
)

A delete needs only the key:

await drasi.push_change("orders", {"op": "delete", "id": "o1", "labels": ["Order"]})

Relations

Supplying start_id and end_id makes the change a relation rather than a node:

await drasi.push_change(
    "graph",
    {"op": "insert", "id": "r1", "labels": ["OWNS"], "start_id": "c1", "end_id": "o1"},
)

The Node.js spellings startId/endId and inId/outId are accepted too, so a change built for @drasi/lib can be passed straight through.

Both endpoints must exist before the relation is pushed, or the change is rejected.

Validation happens early

A malformed change raises immediately, on the call rather than the await:

pending = drasi.push_change("orders", {"id": "o1"})  # raises here

The error carries a code you can branch on — CHANGE_OP_REQUIRED in that case. See error handling.

Concurrency

push_change is safe to call concurrently on the same source. Dispatch is serialised internally, so changes reach queries in the order their sequence numbers were assigned:

await asyncio.gather(*(drasi.push_change("orders", order(i)) for i in range(100)))

Ordering with respect to queries

A query only sees changes that arrive after it subscribes, so push after the query is running:

await drasi.add_query("open", OPEN_ORDERS, ["orders"])
await drasi.wait_for_query("open")
await drasi.push_change("orders", ...)

For a source backed by a real system, use a bootstrap provider instead, which loads what already exists.

Next

• Python reactions — receiving the diffs.
• Streaming results — consuming them as an async iterator.