walbox: react to PostgreSQL changes from Python

[u/mochama254]

I built this because I wanted to react to PostgreSQL changes from Python without polling, without triggers, and without pulling in a whole CDC platform.

It consumes PostgreSQL logical replication and exposes committed transactions as an async stream in Python.

What it does:

• Keeps a durable checkpoint. If the process dies, it resumes from the last transaction it actually finished, not the last one it started.
• Bounded delivery queue, so a slow handler doesn't let memory grow without limit.
• Reconnects automatically after the connection drops.
• One dependency: psycopg3.

The transactional outbox is one use case, but it works with any published table.

GitHub: https://github.com/mochams/walbox

Curious to hear where this wouldn't fit your setup, or what's missing if you've solved this problem a different way.

[u/memeface231]

This must be amazing for some people. I just setup views which trigger celery tasks to handle changes, even works for bulk inserts and updates which don't trigger signals. Many ways to Rome right?

[u/mochama254 → u/[deleted]]

Good questions, thanks for digging in.

Failover / multiple instances

walbox is single-consumer per replication slot on purpose. It opens one replication connection for the slot you configure and reconnects if that connection drops, but it doesn't do leader election.

That said, PostgreSQL's slot exclusivity gets you further than it looks. Only one connection can hold START_REPLICATION on a slot at a time, so you can point two instances at the same slot_name. One gets accepted and starts consuming. The other gets rejected and sits in walbox's reconnect loop until the first connection goes away. That's a working warm standby without adding any coordination service, and PostgreSQL remains the sole arbiter of which instance owns the replication stream.

The rough edge right now is the reconnect backoff. It's capped at 60 seconds and isn't configurable. I'm considering making that configurable and writing up the standby pattern properly. I’m not planning to build general leader election into walbox; that remains an application or infrastructure concern.

Publication sync

walbox only creates or reuses the replication slot. It never creates or alters the publication. So when a Django migration adds a table, that table won't show up in the stream until the publication is updated too.

In a Django app, the cleanest way to handle this is to tie the publication change to the migration that adds the table:

def create_publication(apps, schema_editor):
    EventLog = apps.get_model('data', 'EventLog')
    table = EventLog._meta.db_table
    publication = schema_editor.quote_name(settings.WALBOX_PUBLICATION_NAME)
    schema_editor.execute(f'CREATE PUBLICATION {publication} FOR TABLE {table}')


def drop_publication(apps, schema_editor):
    publication = schema_editor.quote_name(settings.WALBOX_PUBLICATION_NAME)
    schema_editor.execute(f'DROP PUBLICATION IF EXISTS {publication}')


class Migration(migrations.Migration):

    dependencies = [
        ('...', '000...'),
    ]

    operations = [
        migrations.RunPython(create_publication, drop_publication),
    ]

That way the publication change ships with the schema change instead of being a separate manual step someone has to remember.

Both of these boundaries are deliberate. walbox handles consuming the stream and resuming it reliably. Topology, ownership, and publication policy stay with the application and the infrastructure around it.

Appreciate you flagging both.

A few other things worth checking in a Django app

A couple of Django system checks catch the most common setup mistakes before they turn into a confusing runtime error:

@register()
def check_wal_level(
    app_configs: Sequence[AppConfig] | None,
    **kwargs: Any,
) -> list[Warning]:
    """Check if the PostgreSQL WAL level is set to 'logical' for replication.

    Returns:
        list[Warning]: A warning if the WAL level is not 'logical'.
    """
    with connection.cursor() as cursor:
        cursor.execute("SHOW wal_level;")
        wal_level = cursor.fetchone()[0]

    if wal_level != "logical":
        return [
            Warning(
                f"wal_level is '{wal_level}', but logical replication "
                "requires 'logical'.",
                hint="Set wal_level to 'logical' in your PostgreSQL configuration"
                " and restart the PostgreSQL server.",
                id="replication.E001",
            ),
        ]
    return []


@register()
def check_publication(
    app_configs: Sequence[AppConfig] | None,
    **kwargs: Any,
) -> list[Warning]:
    """Check if the PostgreSQL publication is set up for replication.

    Returns:
        list[Warning]: A warning if the publication is not set up.
    """
    publication_name = settings.WALBOX_PUBLICATION_NAME
    with connection.cursor() as cursor:
        cursor.execute(
            "SELECT 1 FROM pg_publication WHERE pubname = %s",
            [publication_name],
        )
        exists = cursor.fetchone() is not None

    if not exists:
        return [
            Warning(
                f"Publication '{publication_name}' does not exist.",
                hint=(
                    "Create it with a migration "
                    f"(CREATE PUBLICATION {publication_name} FOR TABLE ...) "
                    "or manually in your PostgreSQL database."
                ),
                id="replication.E002",
            ),
        ]
    return []


@register()
def check_replication_role(
    app_configs: Sequence[AppConfig] | None,
    **kwargs: Any,
) -> list[Warning]:
    """Check if the PostgreSQL replication role is set up for replication.

    Returns:
        list[Warning]: A warning if the connecting role can't replicate.
    """
    db = settings.DATABASES["default"]
    user = db["USER"]

    with connection.cursor() as cursor:
        cursor.execute(
            "SELECT rolreplication OR rolsuper FROM pg_roles WHERE rolname = %s",
            [user],
        )
        (has_replication,) = cursor.fetchone()

    if not has_replication:
        return [
            Warning(
                "The connecting role cannot open a replication connection.",
                hint=(
                    "Grant replication privileges to the user "
                    f"with a migration (ALTER ROLE {user} WITH REPLICATION) "
                    "or manually in your PostgreSQL database."
                ),
                id="replication.E003",
            ),
        ]
    return []