quickstart.py

"""Minimal, runnable example — no external dependencies.

Run it:

    python examples/minimal/quickstart.py

It builds a CrudHandlers over the in-memory store (the default seam) and shows
the whole entity-agnostic CRUD surface for a generic "widgets" entity:

  1. create / get / patch / delete a row, with a per-mutation audit seam;
  2. tenant isolation — one tenant's rows are invisible to another;
  3. filter + keyset cursor pagination (stable under concurrent inserts);
  4. a bulk update across many rows in one call.

Everything here is pure standard library and runs on Python 3.9. The same
:class:`EntitySpec` drives a FastAPI router via ``build_crud_router`` (see
docs/FORKING.md) and the clean-room Vue/Pinia store under ``frontend/``.
"""
import asyncio
import json

from vue_pinia_crud_store import CrudHandlers, InMemoryStore, NotFound, WIDGET_SPEC


async def main() -> None:
    # An audit seam that just prints — in production this writes an append-only
    # ledger row. Mutations never fail if it raises (best-effort by design).
    audit_trail = []

    async def audit(tenant_id, **kw):
        audit_trail.append((tenant_id, kw["operation"], kw.get("entity_id")))

    store = InMemoryStore(pk_column="id")
    widgets = CrudHandlers(WIDGET_SPEC, store, audit_emitter=audit)

    print("# 1. create / get / patch / delete (with per-mutation audit)")
    row = await widgets.create("tenant-A", {"name": "alpha", "status": "active", "quantity": 3})
    wid = row["id"]
    print("created:", json.dumps({k: row[k] for k in ("id", "name", "status")}))

    fetched = await widgets.get("tenant-A", wid)
    assert fetched["name"] == "alpha"

    patched = await widgets.patch("tenant-A", wid, {"status": "archived"})
    print("patched status ->", patched["status"])
    assert patched["status"] == "archived"

    print("\n# 2. tenant isolation — tenant-B cannot see tenant-A's row")
    await widgets.create("tenant-B", {"name": "beta"})
    try:
        await widgets.get("tenant-B", wid)
        raise AssertionError("tenant-B should not see tenant-A's row")
    except NotFound:
        print("tenant-B got NotFound for tenant-A's row (correct)")

    print("\n# 3. filter + keyset cursor pagination")
    for i in range(5):
        await widgets.create("tenant-A", {"name": "w%d" % i, "status": "active"})
    page1 = await widgets.list("tenant-A", limit=2)
    print("page 1 size:", len(page1["widgets"]), "| has next:", page1["next_cursor"] is not None)
    page2 = await widgets.list("tenant-A", limit=2, cursor=page1["next_cursor"])
    print("page 2 size:", len(page2["widgets"]))
    ids1 = {w["id"] for w in page1["widgets"]}
    ids2 = {w["id"] for w in page2["widgets"]}
    assert ids1.isdisjoint(ids2), "pages must not overlap"

    active = await widgets.list(
        "tenant-A", filters=[{"field": "status", "op": "eq", "value": "active"}]
    )
    print("active widgets:", len(active["widgets"]))

    print("\n# 4. bulk update across many rows in one call")
    all_ids = [w["id"] for w in (await widgets.list("tenant-A", limit=200))["widgets"]]
    result = await widgets.bulk(
        "tenant-A",
        {"operation": "update", "ids": all_ids, "patch": {"status": "done"}},
    )
    print("bulk affected:", result["affected"])

    await widgets.delete("tenant-A", wid)

    print("\n# audit trail (one row per mutation):")
    for entry in audit_trail:
        print("  ", entry)


if __name__ == "__main__":
    asyncio.run(main())