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
pip install "grampy-q[postgres]"  # + the PostgreSQL drivers (SQLAlchemy)

The distribution is grampy-q — the q is for queue, grampy being taken on PyPI — and it is imported as quazardous.grampy.

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 dataclasses import dataclass

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

@dataclass
class Order:                                   # your object, as it is
    id: int
    total: float

class Orders(Adapter):
    def id_of(self, order):  return order.id   # the one thing grampy must know

GRAPH = (Node("pay"), Node("ship", parents=("pay",)))
items = Items(NodeJournal(MemoryDriver(), GRAPH), Orders())
orders = [Order(1, 30.0), Order(2, 12.5)]

lease = items.claim("pay", 10, candidates=orders)       # orders in, orders out
for order in lease:
    print("paying", order.id, order.total)
items.conclude("pay", lease)

print(items.progress(orders[0]))                        # {'pay': Status.DONE}
print(list(items.claim("ship", 10, candidates=orders)))  # pay is done: ship is next

That runs as pasted. A larger graph, with subjects that differ:

from dataclasses import dataclass

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

DAG = (
    # `working` and `state`: what YOUR status column says while the node runs
    # and once it is done — the journal records `running` and `done`.
    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)

@dataclass(frozen=True)
class Doc:
    id: str
    scanned: bool = True

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

def my_loader():                              # your query, your ORM
    return [Doc("s1"), Doc("s2", scanned=False)]

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)

working and state are projections, not commands: allowed_transitions() turns them into the transitions your own status column may take — see the rules.

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.

Why not a status column?

A status column and a few UPDATEs hold one line of steps, one worker at a time. What grampy adds, each proven by the shared driver contract (quazardous.grampy.testing.JournalContract):

• Two workers, one subject. A claim is atomic and returns a token; a slow worker whose lease went to another writes nothing (test_concurrent_claimers_never_take_a_subject_twice, test_a_released_lease_cannot_conclude_the_next_one).
• A graph, not a line. Forks run in parallel, joins wait for their parents, optional steps and exclusive choices do not block what follows (test_the_driver_claims_exactly_what_the_rule_says, test_a_child_waits_for_its_parent_to_conclude).
• Going back. Forget, replay and loops keep what they take away, with when and why (test_forget_and_release_archive_what_they_take_away).
• Time. Retries, leases, waits and grace periods, settled by one janitor call (test_a_failure_is_retried_when_due_then_stands).

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.

Next to other tools

grampy is a journal, not a runner: the comparison is about where the state lives. Each cell links the page that says so; — is a question that page does not settle. Checked on 2026-09-18.

How each notion maps, tool by tool: the same notions in other tools.

Documentation

• The rules — every mechanism, spelled out.
• Items — objects instead of ids, the canonical way.
• Drivers and candidates — tables, queries, your own data.
• Writing a driver — the capabilities, what each method may return, certifying with the contract.
• Operating in production — the indexes worth adding, measured; what grows; the janitor.
• 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