In my django application I have a list of notification types and I want to allow customers to subscribe to one or more notification types.

Each notification type has somewhat of a custom logic so the code of each notification has to be in a different class but I have created a singleton class that gathers all the notification type names and other metadata like description etc.

I want to have a list in the database of all the supported notification types so that the relationship between customers and notification types can be stored in the database while customers subscribe to notification types. I want to have a notification type table so that I can store the metadata and a separate table to store the many-to-many relationship between customers and notifications.

That is where connection_created signal comes in. I have created the following signal that creates the notification type items in the database when the connection_created signal is received so they get auto-updated when I am changing the code:

from django.db.backends.signals import connection_created
from django.db.backends.sqlite3.base import (
    DatabaseWrapper as SQLiteDatabaseWrapper,
)
from django.db.backends.postgresql.base import (
    DatabaseWrapper as PostgresDatabaseWrapper,
)
from django.dispatch import receiver

from notification_type_singleton import NotificationTypeSingleton
from models import NotificationType


@receiver(connection_created, sender=SQLiteDatabaseWrapper)
@receiver(connection_created, sender=PostgresDatabaseWrapper)
def create_or_update_notification_type(sender, **kwargs):
    exiting_ids = []
    for _, notification_type in (
        NotificationTypeSingleton._data.items()
    ):
        notification, _ = NotificationType.objects.update_or_create(
            name=notification_type.name,
            defaults={
                'description': notification_type.description,
                'is_active': True,
            },
        )
        exiting_ids.append(notification.id)

    # Deactivate all notifications in the database that are not used
    NotificationType.objects.exclude(id__in=exiting_ids).update(is_active=False)

    # Update the registry with the created events
    NotificationTypeSingleton._registry = {
        notification.name: notification
        for notification in NotificationType.objects.filter(is_active=True)
    }

That seems to work fine when I bring up my application with python manage.py runserver but when I test with pytest and postgres, the signal is raised as expected but I get RuntimeWarning: Normally Django will use a connection to the 'postgres' database to avoid running initialization queries against the production database when it's not needed (for example, when running tests). Django was unable to create a connection to the 'postgres' database and will use the first PostgreSQL database instead. and if I comment out the code that is doing the queries in the signal then the error goes away.

At the same time according to the docs the connection_created signal should be used for

...any post connection commands to the SQL backend.

so I think I am in the intended use cases but it seems that there are side effects? Does anybody else have had similar problems with testing?

I think this is a common issue when using the connection_created signal in tests. It happens for several reasons, such as:

• Early firing so connection_created runs very early in Django’s DB setup.

• Test DB creation so when pytest runs, Django first connects to the postgres admin DB to spin up test DBs.

• ORM queries too soon so your handler tries to run NotificationType.objects.update_or_create() before the app DB is fully ready.

To fix that, you can use, post_migrate, something like this:

from django.db.models.signals import post_migrate
from django.dispatch import receiver
from notification_type_singleton import NotificationTypeSingleton
from models import NotificationType

@receiver(post_migrate)
def create_or_update_notification_types(sender, **kwargs):
    # Only run for your app to avoid running multiple times !!!!!!!
    if sender.name != 'your_app_name':
        return
        
    existing_ids = []
    for _, notification_type in NotificationTypeSingleton._data.items():
        notification, _ = NotificationType.objects.update_or_create(
            name=notification_type.name,
            defaults={
                'description': notification_type.description,
                'is_active': True,
            },
        )
        existing_ids.append(notification.id)

    # Deactivate all notifications in the database that are not used

    NotificationType.objects.exclude(id__in=existing_ids).update(is_active=False)

    # Update the registry with the created events
    NotificationTypeSingleton._registry = {
        notification.name: notification
        for notification in NotificationType.objects.filter(is_active=True)
    }

Update:

The post_migrate signal only runs when migrations are executed, not during regular runserver operations. So, I suggest to you to combine post_migrate with ready() (method for runserver), something like this, and don't forget to change 'your_app_notificationtype' with your actual table name and YourAppConfig (in your app.py)

# apps.py
from django.apps import AppConfig
from django.db.models.signals import post_migrate

class YourAppConfig(AppConfig):
    default_auto_field = 'django.db.models.BigAutoField'
    name = 'your_app_name'

    def ready(self):
        # Import here to avoid circular imports
        from django.db import connection
        from django.core.management.color import no_style
        
        # Check if tables exist before trying to access them
        if self._tables_exist():
            self._initialize_notification_types()
        
        # Also register the post_migrate signal
        post_migrate.connect(self._post_migrate_handler, sender=self)

    def _tables_exist(self):
        """Check if the required tables exist"""
        from django.db import connection
        table_names = connection.introspection.table_names()
        return 'your_app_notificationtype' in table_names

    def _initialize_notification_types(self):
        """Initialize notification types - shared logic"""
        from .models import NotificationType
        from .notification_type_singleton import NotificationTypeSingleton
        
        existing_ids = []
        for _, notification_type in NotificationTypeSingleton._data.items():
            notification, _ = NotificationType.objects.update_or_create(
                name=notification_type.name,
                defaults={
                    'description': notification_type.description,
                    'is_active': True,
                },
            )
            existing_ids.append(notification.id)

        # Deactivate unused notifications
        NotificationType.objects.exclude(id__in=existing_ids).update(is_active=False)

        # Update the registry
        NotificationTypeSingleton._registry = {
            notification.name: notification
            for notification in NotificationType.objects.filter(is_active=True)
        }

    def _post_migrate_handler(self, sender, **kwargs):
        """Handle post_migrate signal"""
        if sender.name == self.name:
            self._initialize_notification_types()