Examples

The scripts in examples/ run as they are against the compose stack (docker compose --profile clickhouse up -d --wait), and the test suite runs every one of them, so what is shown here works. Point them at other stores with RHIZODAT_PG, RHIZODAT_CLICKHOUSE, RHIZODAT_GRAPH and RHIZODAT_JOURNAL.

One entity across two stores

"""Quickstart: one entity, two stores, every version kept.

An order's money lives in PostgreSQL and its links in the drevo graph. Each
write is one atomic meta-transaction over both; reads see any past moment.
"""

import time
from datetime import UTC, datetime
from decimal import Decimal

from _stores import GRAPH, JOURNAL, PG

from rhizodat import Rhizodat, entity


@entity(stores={"pg": ["status", "amount"], "graph": ["customer"]}, name="shop.Order")
class Order:
    status: str
    amount: Decimal
    customer: str


with Rhizodat(stores={"pg": PG, "graph": GRAPH}, journal=JOURNAL) as dat:
    orders = dat.repository(Order)

    order = Order(status="new", amount=Decimal("10.00"), customer="ada")
    orders.add(order)  # one meta-transaction across PostgreSQL and drevo
    before = datetime.now(UTC)
    time.sleep(0.01)  # the update must come after `before`

    order.status = "paid"
    orders.update(order)

    print(orders.get(order.dat_id).status)  # paid
    print(orders.get(order.dat_id, as_of=before).status)  # new
    print([v.status for v in orders.history(order.dat_id)])  # ['new', 'paid']

A Unit of Work with native statements

"""A Unit of Work with native statements in two stores, re-run on transient errors.

A worker claims a task in PostgreSQL and marks itself busy in the graph: both
or neither. ``transactional`` re-runs the whole function with a new
meta-transaction when a store reports a transient failure.
"""

from _stores import GRAPH, JOURNAL, PG

from rhizodat import Retry, Rhizodat, UnitOfWork

with Rhizodat(stores={"pg": PG, "graph": GRAPH}, journal=JOURNAL) as dat:
    dat.store("pg").execute(
        "CREATE TABLE IF NOT EXISTS example_task (id serial PRIMARY KEY, claimed_by text)"
    )
    dat.store("pg").execute("INSERT INTO example_task (claimed_by) VALUES (NULL)")
    dat.store("graph").execute(
        "MERGE (w:ExampleWorker {title: 'ExampleWorker:ada', name: 'ada'}) SET w.busy = false"
    )

    @dat.transactional(retry=Retry(attempts=5, deadline_s=10))
    def claim(uow: UnitOfWork, worker: str) -> int:
        pg = uow.store("pg")
        row = pg.execute(
            "SELECT id FROM example_task WHERE claimed_by IS NULL ORDER BY id LIMIT 1"
        ).one()
        pg.execute("UPDATE example_task SET claimed_by = %s WHERE id = %s", (worker, row["id"]))
        uow.store("graph").execute(
            "MATCH (w:ExampleWorker {name: $w}) SET w.busy = true", {"w": worker}
        )
        return int(row["id"])

    outcome = claim("ada")  # Committed (or CommittedPending); raises Aborted on failure
    print(f"claimed task {outcome.value}")

Valid time and transaction time

"""Valid time and transaction time: what held when, as known when.

A price agreement holds for a period; a rebate entered later wins for its own
days only, and the price as known before the rebate is still readable.
"""

import time
from datetime import UTC, datetime, timedelta
from decimal import Decimal

from _stores import JOURNAL, PG

from rhizodat import Rhizodat, entity


@entity(stores={"pg": ["customer", "price"]}, name="erp.Price", valid_time=True)
class Price:
    customer: str
    price: Decimal


def day(n: int) -> datetime:
    return datetime(2026, 1, 1, tzinfo=UTC) + timedelta(days=n)


with Rhizodat(stores={"pg": PG}, journal=JOURNAL) as dat:
    prices = dat.repository(Price)
    p = Price(customer="acme", price=Decimal("10.00"), valid_from=day(0), valid_to=None)
    prices.add(p)
    before_rebate = datetime.now(UTC)
    time.sleep(0.01)

    p.price, p.valid_from, p.valid_to = Decimal("8.00"), day(10), day(20)  # a rebate
    prices.update(p)

    def price(at: datetime, known_at: datetime | None = None) -> Decimal:
        found = prices.get(p.dat_id, as_of=at, known_at=known_at)
        assert found is not None
        return found.price

    print(price(day(5)), price(day(15)), price(day(25)))  # 10.00 8.00 10.00
    print(price(day(15), known_at=before_rebate))  # 10.00: as known before the rebate

Schema migrations

examples/migrations/0001_tasks.toml:

description = "tasks in PostgreSQL, their activity log in ClickHouse"

[[steps]]
store = "pg"
sql = """
CREATE TABLE IF NOT EXISTS example_tracker_task (
    id uuid PRIMARY KEY,
    title text NOT NULL
)"""

[[steps]]
store = "events"
sql = "CREATE TABLE IF NOT EXISTS example_activity (task UUID, at DateTime64(3), what String) ENGINE = MergeTree ORDER BY (task, at)"

"""Schema migrations: one history for every store, kept in the journal.

The directory ``examples/migrations`` runs from code, or with
``rhizodat migrate examples/migrations``.
A second run applies nothing; ``schema_versions`` says what applied when.
"""

from pathlib import Path

from _stores import CLICKHOUSE, JOURNAL, PG

from rhizodat import Rhizodat

with Rhizodat(stores={"pg": PG, "events": CLICKHOUSE}, journal=JOURNAL) as dat:
    report = dat.migrate(Path(__file__).parent / "migrations")
    print(f"applied {report.applied}, already {report.already}")
    for v in dat.schema_versions():
        print(v.id, v.name, v.applied_at.isoformat())

Importing a CSV with its schema inferred

"""Import a CSV into ClickHouse with its schema inferred from every row.

The new table becomes a migration file next to the others (here a temporary
directory), runs, and the rows load as one unit. The same from a shell:
``rhizodat import sales.csv --store events --table sales --migrations migrations``.
"""

import random
import tempfile
from pathlib import Path

from _stores import CLICKHOUSE

from rhizodat import Rhizodat
from rhizodat.ingest import import_file

tmp = tempfile.mkdtemp()
# A journal of its own: the migration history belongs to this example's directory.
with Rhizodat(stores={"events": CLICKHOUSE}, journal=f"file://{tmp}/journal.db") as dat:
    sales = Path(tmp) / "sales.csv"
    rows = ["order_id,amount,placed_at,channel"]
    for i in range(1, 101):
        channel = random.choice(["web", "shop", "phone"])
        amount = random.randint(100, 99999) / 100
        rows.append(f"{i},{amount},2026-10-{i % 28 + 1:02} 12:00:00,{channel}")
    sales.write_text("\n".join(rows) + "\n")

    migrations = Path(tmp) / "migrations"
    migrations.mkdir()
    report = import_file(
        dat,
        sales,
        store="events",
        table="example_sales",
        migrations=migrations,
        order_by=["order_id"],
        replace=True,
    )
    print(f"{report.rows} rows, migration {report.migration}")
    for c in report.plan.schema.columns:
        print(f"  {c.name}: {c.type}{' (low cardinality)' if c.low_cardinality else ''}")