grampy

A small workflow graph for work queues that already live in a storage — a database, a key-value store, or plain memory: the storage is a driver.

pip install grampy-q              # from quazardous import grampy
pip install "grampy-q[postgres]"  # + the PostgreSQL driver (SQLAlchemy)

Try the brick sorter → — a sorting line where bricks are sorted by colour, TNT bricks get quarantined, defused with retries or thrown away, and a sorted brick sent back waits in a lane before it runs again. It runs the real library in your browser.

You declare a DAG of nodes. Each subject (a job, a request, an offer…) goes through the nodes; a node journal records, per subject and per node, whether the node is running, done, skipped, failed or omitted. Workers claim a node for eligible subjects, conclude it, and the graph decides what becomes claimable next — forks run in parallel, joins wait for their parents.

from quazardous.grampy import Node, NodeJournal, check_dag
from quazardous.grampy.drivers.memory import MemoryDriver
from quazardous.grampy.items import Adapter, Items

DAG = (
    Node("fetch", working="fetching", state="fetched"),
    Node("crop", parents=("fetch",), optional=True),
    Node("read", parents=("fetch",), optional=True),
    Node("judge", parents=("crop", "read"), state="judged"),
)
check_dag(DAG)

class DocsAdapter(Adapter):                       # how grampy reads YOUR object
    def id_of(self, doc):  return doc.id
    def inflate(self, ids):   return [LIBRARY[i] for i in ids]
    def applies(self, doc, node):          # one graph, subjects that differ
        return node != "crop" or doc.scanned

items = Items(NodeJournal(MemoryDriver(), DAG), DocsAdapter())
lease = items.claim("fetch", 10, candidates=my_loader())    # objects in…
for doc in lease:                                           # …and objects out
    ...
items.conclude("fetch", lease)

Handing grampy your objects is the canonical way to use it — see items. The core underneath works on ids alone and stays available: journal.claim("fetch", 10, candidates=[…]) returns subjects, and never reads your data.

The package lives in the quazardous namespace; the distribution is grampy-q (grampy was already taken on PyPI; the q is for queue).

What it does

Each line links to the rules, where it is spelled out.

Storage

The journal never commits and never reads your tables. Eligibility is your query, passed to claim and read inside its transaction; a claim gives back ids, and you load your own objects. See drivers.

Any other storage: implement the driver protocol and pass the shared contract (quazardous.grampy.testing.JournalContract), concurrency tests included.

Documentation

• The rules — every mechanism, spelled out.
• Items — objects instead of ids, the canonical way.
• Drivers and candidates — tables, queries, your own data.
• grampy's tables next to yours — what to read, what never to write.
• Drawings — diagrams from a graph.
• The same notions in other tools — Graphile Worker, BullMQ, Hatchet, Inngest, Temporal, Oban.
• CONTRIBUTING — tests, lint, how to release.

Tests

pip install -e ".[test]"        # or: uvx --with pytest pytest
pytest                          # graph, states, memory and SQLite drivers
GRAMPY_TEST_PG_DSN=postgresql+psycopg://user:pass@localhost/test \
    pytest                      # + the postgres driver (needs sqlalchemy and a driver)

License

MIT