Why Gravtory?

Every production application eventually needs workflows that cannot fail halfway. A payment that must complete. An order that must ship. A data pipeline that must finish. When a process crashes between steps, you need guarantees.

The current options each carry significant trade-offs:

• Celery accepts that tasks can be lost on crash.
• Temporal requires a multi-service deployment and a steep learning curve.
• Prefect Cloud and Airflow demand hosted infrastructure or managed clusters.
• Hand-rolled retry loops provide no real durability.

Gravtory offers an alternative: Temporal-grade reliability delivered as a Python library, using the database you already operate.

pip install gravtory[postgres]

That is the entire infrastructure requirement.

Quick Start

1. Define a workflow

from gravtory import Gravtory, workflow, step

grav = Gravtory("postgresql://localhost/mydb")

@grav.workflow(id="order-{order_id}")
class OrderWorkflow:

    @step(1)
    async def charge_card(self, order_id: str) -> dict:
        return await stripe.charge(order_id)

    @step(2, depends_on=1)
    async def reserve_inventory(self, order_id: str) -> dict:
        return await inventory.reserve(order_id)

    @step(3, depends_on=2)
    async def send_notification(self, order_id: str) -> None:
        await email.send(order_id)

2. Run it

await grav.start()
result = await grav.run(OrderWorkflow, order_id="ord_abc123")

3. It survives anything

Normal run:   step 1 [OK]   -> step 2 [OK]   -> step 3 [OK]     completes
Crash:        step 1 [OK]   -> step 2 [CRASH]                    process dies
Auto-resume:  step 1 [SKIP] -> step 2 [OK]   -> step 3 [OK]     resumes exactly

Step 1 (charge card) is never re-executed. Its output was atomically checkpointed to the database. On restart, Gravtory loads that checkpoint and continues from the precise point of failure.

Features

Comparison

Patterns

Saga with Automatic Compensation

@grav.workflow(id="transfer-{id}")
@saga
class TransferWorkflow:

    @step(1, compensate="refund")
    async def debit(self, amount: Decimal) -> dict:
        return await bank.debit(self.source, amount)

    @step(2, depends_on=1, compensate="reverse")
    async def credit(self, amount: Decimal) -> dict:
        return await bank.credit(self.dest, amount)

    async def refund(self, output: dict):
        await bank.credit(self.source, output["amount"])

    async def reverse(self, output: dict):
        await bank.reverse(output["transaction_id"])

# If credit fails, refund runs automatically -- crash-safe.

Parallel Processing

@grav.workflow(id="batch-{id}")
class BatchWorkflow:

    @step(1)
    async def get_items(self, id: str) -> list[str]:
        return await db.get_item_ids(id)

    @step(2, depends_on=1)
    @parallel(max_concurrency=20)
    async def process(self, item_id: str) -> dict:
        return await compute(item_id)
    # Each item is individually checkpointed.
    # On crash: only unfinished items re-execute.

    @step(3, depends_on=2)
    async def summarize(self, results: list[dict]) -> dict:
        return {"processed": len(results)}

Human-in-the-Loop

@grav.workflow(id="expense-{id}")
class ExpenseWorkflow:

    @step(1)
    async def submit(self, id: str, amount: float) -> dict:
        await slack.send(f"Approve expense #{id} (${amount})?")
        return {"id": id, "amount": amount}

    @step(2, depends_on=1)
    @wait_for_signal("approval", timeout=timedelta(days=7))
    async def await_approval(self, signal: dict) -> bool:
        return signal["approved"]

    @step(3, depends_on=2, condition=lambda ctx: ctx.output(2))
    async def reimburse(self, id: str) -> None:
        await accounting.pay(id)

# From your API or Slack bot:
await grav.signal("expense-42", "approval", {"approved": True})

Scheduled Workflows

@grav.workflow(id="daily-report")
@grav.schedule(cron="0 9 * * *", tz="US/Eastern")
class DailyReport:

    @step(1)
    async def generate(self) -> dict:
        return await analytics.report()

    @step(2, depends_on=1)
    async def send(self, report: dict) -> None:
        await email.send_report(report)

Retry with Backoff

@step(1, retries=5, backoff="exponential", backoff_base=2.0, retry_on=[httpx.TimeoutError])
async def call_external_api(self, url: str) -> dict:
    return await httpx.get(url).json()
# Retries at: 2s, 4s, 8s, 16s, 32s (with jitter)

Distribution

No message broker required. Workers coordinate through the database.

# Scale from 1 to N workers on a single machine
grav = Gravtory("postgresql://localhost/mydb", workers=8)

# Scale across machines -- same code, different hosts
# Machine A:
grav = Gravtory("postgresql://shared-db/workflows", workers=8, node_id="a")
# Machine B:
grav = Gravtory("postgresql://shared-db/workflows", workers=8, node_id="b")

Observability

grav = Gravtory(
    "postgresql://localhost/mydb",
    dashboard=True,              # Web UI at :7777
    otel_endpoint="jaeger:4317", # OpenTelemetry traces
    metrics_port=9090,           # Prometheus metrics
)

# Introspection API
state = await grav.inspect("order-ord_123")
print(state.status)            # "completed"
print(state.steps[1].output)   # {"charge_id": "ch_xyz"}
print(state.steps[1].duration_ms)  # 142

# Failure hooks
@grav.on_failure
async def alert(ctx):
    await slack.send(f"Workflow {ctx.workflow_run_id} failed: {ctx.error}")

Testing

No database required. The built-in test runner operates entirely in memory.

from gravtory.testing import WorkflowTestRunner

async def test_order_workflow():
    runner = WorkflowTestRunner()  # In-memory
    runner.mock(OrderWorkflow.charge_card, return_value={"charge_id": "test"})
    runner.mock(OrderWorkflow.reserve_inventory, return_value={"ok": True})
    runner.mock(OrderWorkflow.send_notification, return_value=None)

    result = await runner.run(OrderWorkflow, order_id="test_123")
    assert result.status == "completed"

    # Simulate crash and verify resume
    runner.simulate_crash_after(step=1)
    result = await runner.run(OrderWorkflow, order_id="test_456")
    result = await runner.resume("order-test_456")
    assert result.steps[1].was_replayed  # Not re-executed

CLI

gravtory list --status=failed
gravtory inspect order-ord_123
gravtory retry order-ord_123
gravtory signal expense-42 approval '{"approved": true}'
gravtory dlq list
gravtory dashboard
gravtory workers start --count=4

Installation

# Core + PostgreSQL (recommended for production)
pip install gravtory[postgres]

# Core + SQLite (local development)
pip install gravtory[sqlite]

# Core + MySQL
pip install gravtory[mysql]

# Core + MongoDB
pip install gravtory[mongodb]

# All backends and optional extras
pip install gravtory[all]

Requirements: Python 3.10+

Gravtory ships with a py.typed marker (PEP 561). Full type annotations work out of the box with mypy, pyright, and other type checkers.

Backends

Framework Integration

FastAPI

from contextlib import asynccontextmanager

from fastapi import FastAPI
from gravtory import Gravtory

grav = Gravtory("postgresql://localhost/mydb")

@asynccontextmanager
async def lifespan(app):
    await grav.start()
    yield
    await grav.shutdown()

app = FastAPI(lifespan=lifespan)

@app.post("/orders/{order_id}")
async def create_order(order_id: str):
    run_id = await grav.run(OrderWorkflow, order_id=order_id, background=True)
    return {"run_id": run_id}

Django

Django integration is planned for a future release. Progress is tracked in GitHub Issues.

Coming from Celery?

# Before (Celery)
@app.task(bind=True, max_retries=3)
def charge_card(self, order_id):
    try:
        result = stripe.charge(order_id)
    except Exception as exc:
        raise self.retry(exc=exc)
    return result

# If this crashes: task may be lost. Result may be lost.
# If charge succeeds but ack fails: card charged twice.

# After (Gravtory)
@step(1, retries=3, backoff="exponential")
async def charge_card(self, order_id: str) -> dict:
    return await stripe.charge(order_id)

# If this crashes: checkpoint guarantees at-least-once with idempotent replay.
# On resume: if charge completed, it's loaded from DB. Never re-executed.

License

Gravtory is released under the GNU Affero General Public License v3.0 (AGPL-3.0-or-later).

For commercial licensing inquiries, contact vatryok@protonmail.com.

Contributing

Contributions are welcome. Please read CONTRIBUTING.md for development setup, coding standards, and submission guidelines.

Support

If you find Gravtory useful, consider supporting its continued development on Ko-Fi.