trio-pg-actors

PostgreSQL-backed actor framework for Python — built on Trio.

No Redis. No RabbitMQ. No Celery. Just PostgreSQL and Python.

---

Why?

Most task-queue and actor frameworks bolt on an external broker, add operational complexity, and scatter your system state across multiple datastores. trio-pg-actors turns your existing PostgreSQL database into a reliable, observable message bus — with actors, pub/sub, delayed messages, dead-letter queues, and auto-diff schema migrations, all in one library.

---

Features

🎭 Virtual Actors

Actors are stateless workers that load their state from PostgreSQL before every message and persist it back after. You never think about in-memory state management or node affinity — any worker on any machine can process any message.

📬 Durable Message Queue

Messages are rows in pg_actor_messages. They survive restarts, crashes, and deployments. No message is ever lost once it is INSERTed into the database.

🔔 LISTEN / NOTIFY Wakeup

Workers sleep until PostgreSQL sends a NOTIFY on pg_actor_new_msg. No polling loops. No wasted CPU. New messages wake workers in milliseconds. The listener uses psycopg3's native conn.notifies() async generator — clean, no callbacks.

⏱️ Delayed Messages

Schedule any message to be delivered in the future:

await actor.remind_me(RetryPayment(order_id=42), delay=300)  # 5 minutes later

🔁 Automatic Retries with Exponential Backoff

Failed messages are retried up to max_retries times. Each retry waits min(2^n, 60) seconds. Messages that exhaust retries (or raise ValidationError) go to the dead-letter table automatically.

💀 Dead-Letter Queue

Every message that cannot be processed ends up in pg_actor_dead_letters with the full error text. Easy to inspect, replay, or archive.

🏢 Multi-Tenant Isolation

Each tenant gets its own PostgreSQL schema. The message queue is shared (public.pg_actor_messages), but actor state is isolated per schema (<tenant>.pg_actor_state). Call system.ensure_schema("tenant_xyz") and you're done.

🔄 Auto-Diff Migrations

MigrationManager compares your Python schema dict against the live information_schema and generates JSON migration files — only for what actually changed. Apply them atomically in a transaction. Roll back the last migration with one call.

target = {
    "orders": {"id": "TEXT PRIMARY KEY", "amount": "DOUBLE PRECISION"},
}
await mgr.create_migration("add_orders", target)
await mgr.apply_migrations()

🦥 Lazy Data Migrations

Actor state can evolve across deployments without downtime. Override migrate_state() to transform old state shapes on the fly — the first time an actor processes a message after an upgrade, its state is migrated in the same transaction.

class OrderActor(VirtualActor):
    schema_version: int = 2
    amount: float = 0.0
    currency: str = "USD"  # new in v2

    @classmethod
    def migrate_state(cls, old_version: int, state: dict) -> dict:
        if old_version < 2:
            state.setdefault("currency", "USD")
        return state

🌿 Trio-Native Concurrency

Workers run inside a trio.Nursery. If one worker crashes, Trio's structured concurrency guarantees a clean, predictable shutdown of the entire system — no silent half-dead processes.

asyncio version	trio version
asyncio.Event	trio.Event (recreated per cycle)
asyncio.gather(*workers)	nursery.start_soon(worker) × N
asyncio.wait_for(e, timeout=3)	with trio.move_on_after(3): await e.wait()
asyncpg callback LISTEN	async for notify in conn.notifies()

🧹 Automatic Garbage Collection

A background task periodically deletes old done messages (default: 7-day retention). Configurable, runs in the same nursery as the workers.

🔍 Full SQL Query Builder

AsyncDBOperations provides a safe, injection-proof query builder with:

• Rich where-dict operators: __gt, __lte, __in, __notin, __isnull, __like, __ilike, __neq
• OR / AND grouping
• JSONB operators: {"metadata->>'theme'": "dark"}
• JOIN with collision resolution (prefixed keys + nested sub-dicts)
• FOR UPDATE [SKIP LOCKED] / FOR SHARE row locking
• insert_many / update_many / delete_many
• aggregate and group_by

---

Installation

pip install trio-pg-actors
# or with uv:
uv add trio-pg-actors

Requirements: Python 3.11+, PostgreSQL 14+

---

Quickstart

1. Define messages and actors

# myapp/actors.py
from trio_pg_actors import BaseMessage, VirtualActor, actor, message, subscribe


@message
class CreateOrder(BaseMessage):
    user_id: str
    amount: float


@message
class OrderCreated(BaseMessage):
    order_id: str
    amount: float


@actor
class OrderActor(VirtualActor):
    total_orders: int = 0
    total_revenue: float = 0.0

    async def on_create_order(self, msg: CreateOrder) -> None:
        self.total_orders += 1
        self.total_revenue += msg.amount

        # fan-out an event to all subscribers
        await self.publish(
            target_id=msg.user_id,
            event=OrderCreated(order_id=self._actor_id, amount=msg.amount),
        )


@subscribe(OrderCreated)
@actor
class EmailActor(VirtualActor):
    emails_sent: int = 0

    async def on_order_created(self, msg: OrderCreated) -> None:
        # send confirmation email here
        self.emails_sent += 1

2. Run the system

# main.py
import trio
from trio_pg_actors import PgActorSystem

DSN = "postgresql://user:password@localhost/mydb"


async def main():
    system = PgActorSystem(
        dsn=DSN,
        node_name="worker-1",
        discover_packages=["myapp.actors"],  # auto-imports your actor modules
    )

    async with system.run() as nursery:
        # start 4 worker tasks + GC + LISTEN/NOTIFY listener
        nursery.start_soon(system.run_workers, nursery, 4)

        # send a message — picked up by the next free worker
        await system.tell(
            target_type="OrderActor",
            target_id="order-001",
            msg=CreateOrder(user_id="alice", amount=99.90),
        )

        await trio.sleep_forever()  # run until Ctrl+C


trio.run(main)

3. Multi-tenant example

async def onboard_tenant(system: PgActorSystem, tenant_id: str) -> None:
    # creates schema + isolated pg_actor_state table
    await system.ensure_schema(tenant_id)

    await system.tell(
        target_type="OrderActor",
        target_id="order-001",
        msg=CreateOrder(user_id="bob", amount=49.00),
        schema_name=tenant_id,  # state is isolated to this tenant
    )

---

Project Layout

src/trio_pg_actors/
├── __init__.py      # public API re-exports
├── base_msg.py      # BaseMessage (immutable, UTC timestamp, JSONB helpers)
├── database.py      # AsyncDBOperations + MigrationManager (psycopg3)
└── core.py          # VirtualActor, PgActorSystem, decorators (trio)

---

How It Works

┌─────────────────────────────────────────────────────┐
│                    PostgreSQL                        │
│                                                      │
│  public.pg_actor_messages   ← INSERT (tell/publish)  │
│  public.pg_actor_dead_letters                        │
│  public.pg_actor_state      ← actor state (JSON)    │
│  <tenant>.pg_actor_state    ← isolated per tenant   │
└──────────────────┬──────────────────────────────────┘
                   │  LISTEN / NOTIFY
        ┌──────────▼──────────────────┐
        │       PgActorSystem         │
        │  ┌─────────────────────┐    │
        │  │   Trio Nursery      │    │
        │  │  ├ Worker-0         │    │
        │  │  ├ Worker-1  ───────┼────┤─► SELECT ... FOR UPDATE SKIP LOCKED
        │  │  ├ Worker-N         │    │   dispatch → persist state → mark done
        │  │  ├ GC task          │    │
        │  │  └ Listener task    │    │
        │  └─────────────────────┘    │
        └─────────────────────────────┘

Every worker claims a batch of messages atomically with FOR UPDATE SKIP LOCKED — no message is ever processed twice, even with many workers across many machines.

---

Message Lifecycle

tell() / publish()
      │
      ▼
pg_actor_messages  (status = 'pending')
      │
      ▼  worker claims batch
  dispatch to actor handler
      │
  ┌───┴──────────────────────┐
  │ success                  │ failure
  ▼                          ▼
status = 'done'        retries < max_retries?
                            │ yes → backoff, status stays 'pending'
                            │ no  → pg_actor_dead_letters
                            ▼       status = 'dead'
                      ValidationError → dead immediately

---

License

MIT