django-pgpubsub

In pgpubsub, a listener is a function that processes notifications sent through a specific channel.

To work correctly, listeners must follow these rules:

1. They must be defined in your app's listeners.py file.
2. They must be decorated using decorators from pgpubsub.listen (e.g., @pgpubsub.listener or @pgpubsub.post_insert_listener).
3. The decorator associates the function with a specific channel, which also determines the function's expected signature.
4. The function must be imported in your app's configuration class ready() method (similar to Django signals) to ensure it is registered.

The pgpubsub module uses Channel objects to define the topics for publishing and subscribing.

• pgpubsub.Channel: Represents a standard pub/sub channel.
• pgpubsub.TriggerChannel: A specialized channel type used for triggering specific events.

Notification recovery relies on the lock_notifications setting being enabled on your channel. When enabled, pgpubsub stores a Notification object in the database instead of relying solely on the transient LISTEN/NOTIFY protocol.

If listener processes are offline when a notification is sent, the payload remains in the database. You can then recover these notifications using:

1. The pgpubsub.process_stored_notifications function, which fetches all stored Notifications and sends them to their respective channels.
2. The --recover flag when running the listen management command, which triggers an automatic replay of missed notifications upon startup.

When lock_notifications = True is enabled, pgpubsub changes the notification lifecycle to use database-backed locking:

1. Storage: When a notification is sent (via pgpubsub.notify or a trigger), a pgpubsub.models.Notification object is inserted into the database containing the JSON payload. This is atomic with the Postgres notification.
2. Locking: When a listener detects a notification, it attempts to fetch and lock the corresponding Notification object using select_for_update(skip_locked=True).
3. Concurrency Control: The skip_locked=True flag ensures that if one process has already locked the notification, other processes will immediately skip it rather than waiting. This allows workers to continue polling for other notifications.
4. Cleanup: Once the listener's callback successfully completes, the stored Notification is removed from the database.

Install the library using pip.

Prerequisites:

• Django 2.2 or later.
• PostgreSQL 11 or later.

Setup Steps:

1. Install via pip:

   pip install django-pgpubsub

2. Add pgpubsub and pgtrigger to your INSTALLED_APPS in Django settings.
3. Run python manage.py migrate to create the required Notification model table.

To start listener functions and have them listen for incoming notifications on their associated channels, use the listen management command.

By default, pgpubsub provides automatic fault tolerance: if a listener process encounters a Python-level exception, the library automatically spins up a secondary process to continue listening before the initial process terminates. This prevents the need for manual restarts due to listener exceptions.

./manage.py listen

By default, Postgres LISTEN/NOTIFY notifications are lost if no process is listening when they are sent. To prevent this, you can enable lock_notifications on a channel, which stores a Notification object in the database.

To automatically replay any missed stored notifications when your listener processes start up, use the --recover option with the listen management command.

python manage.py listen --recover

To ensure your listeners are registered when Django starts, you must import the module containing your listeners within the ready() method of your application's AppConfig class.

# tests/apps.py
from django.apps import AppConfig

class TestsConfig(AppConfig):
    name = 'tests'

    def ready(self):
        import pgpubsub.tests.listeners

To ensure your listeners are discovered and registered by pgpubsub, you must import your listeners.py module within the ready() method of your application's AppConfig class.

from django.apps import AppConfig

class TestsConfig(AppConfig):
    name = 'tests'

    def ready(self):
        import pgpubsub.tests.listeners

By default, listener callbacks receive the old and new objects. To also receive the context dictionary used during the trigger, follow these steps:

1. Set PGPUBSUB_PASS_CONTEXT_TO_LISTENERS = True in your Django settings.
2. Add a context parameter (typed as Dict[str, Any]) to your listener function signature.

# listeners.py
import pgpubsub
from typing import Dict, Any

@pgpubsub.post_insert_listener(AuthorTriggerChannel)
def create_first_post_for_author(
    old: Author, 
    new: Author, 
    context: Dict[str, Any]
):
    print(f'Creating first post for {new.name} with context={context}')
    # ... logic ...

# django settings
PGPUBSUB_PASS_CONTEXT_TO_LISTENERS = True