rotor

A durable process runtime for Python.

[!WARNING] Rotor is experimental and unstable. APIs, storage formats, and behavior may change without notice. Do not use it for production workloads.

A process is persisted as a checkpoint and mailbox. A worker leases it for one or more messages, committing each transition independently.

from dataclasses import field
from rotor import DurableProcess, Start, message, on

@message
class Ask:
    question: str

class State:
    asked: list = field(default_factory=list)

class Oracle(DurableProcess[State]):
    @on
    async def start(self, msg: Start):
        self.state.asked.append(f"born knowing: {msg.input}")

    @on
    async def ask(self, msg: Ask):
        self.state.asked.append(msg.question)
        if len(self.state.asked) > 3:
            self.stop(output=self.state.asked)

Timers, child completion, approvals, and limit notifications all arrive as messages. Application code defines their meaning in handlers.

Install

pip install rotorcore              # SQLite tier: zero dependencies
pip install "rotorcore[postgres]"  # + asyncpg for production
pip install "rotorcore[vercel]"    # Queue SDK + Postgres + S3 adapter