Стабильная синхронизация типов уведомлений в Django: post_migrate и AppConfig.ready вместо connection_created

Как в Django стабильно синхронизировать типы уведомлений: избегаем предупреждений в pytest-среде. Вместо connection_created — post_migrate и AppConfig.ready.

Поддерживать эталонный список типов уведомлений в согласованном состоянии между кодом и базой данных кажется обманчиво простым — пока не появляются тесты. Простой хук на уровне подключения вроде бы работает при локальных запусках, но под pytest с PostgreSQL он вызывает шумные предупреждения. Ниже — практический разбор, что именно идет не так и как добиться стабильной работы как в runserver, так и в тестовой среде.

Настройка: синхронизация типов уведомлений при подключении к БД

Цель — хранить в базе реестр поддерживаемых типов уведомлений, поддерживать их метаданные и вести связь многие‑ко‑многим с клиентами. Логика каждого уведомления находится в отдельном классе, а контейнер наподобие синглтона собирает их метаданные. Первая попытка опирается на сигнал connection_created, чтобы синхронизировать строки БД сразу после открытия соединения.

from django.db.backends.signals import connection_created
from django.db.backends.sqlite3.base import DatabaseWrapper as SqliteWrapper
from django.db.backends.postgresql.base import DatabaseWrapper as PgWrapper
from django.dispatch import receiver

from notify_catalog import NotifyCatalog
from models import NoticeType


@receiver(connection_created, sender=SqliteWrapper)
@receiver(connection_created, sender=PgWrapper)
def sync_notice_kinds(sender, **kwargs):
    active_ids = []
    for _, kind in NotifyCatalog._data.items():
        obj, _ = NoticeType.objects.update_or_create(
            name=kind.name,
            defaults={
                'description': kind.description,
                'is_active': True,
            },
        )
        active_ids.append(obj.id)

    # Деактивировать все неиспользуемые уведомления в базе
    NoticeType.objects.exclude(id__in=active_ids).update(is_active=False)

    # Обновить реестр в памяти актуальными событиями
    NotifyCatalog._registry = {
        item.name: item
        for item in NoticeType.objects.filter(is_active=True)
    }

Что на самом деле происходит и почему тесты ругаются

Сигнал connection_created срабатывает очень рано в жизненном цикле работы с базой в Django. Во время тестов Django сначала обращается к административной базе PostgreSQL, чтобы подготовить тестовые базы. Если в этот момент выполнять записи на уровне приложения через ORM, они попадают раньше, чем основная база приложения станет полностью готовой к обычным операциям. Этот сдвиг по времени и проявляется как RuntimeWarning о том, что Django при подготовке тестов откатывается к первой базе PostgreSQL. Иными словами, хук подходит для низкоуровневых настроек соединения, но становится хрупким, если использовать его для начального наполнения или изменения таблиц приложения, пока тестовая инфраструктура ещё загружается.

Стабильное решение: отложить наполнение до post_migrate и покрыть runserver через AppConfig.ready

Перенос синхронизации на более позднюю фазу жизненного цикла устраняет гонку с созданием тестовой базы. Сигнал post_migrate срабатывает после применения миграций — в самый подходящий момент, чтобы вставлять или обновлять строки с метаданными. Для обычных запусков сервера ту же инициализацию можно выполнять из AppConfig.ready, предварительно проверяя наличие таблиц, чтобы не ломаться при первом запуске.

from django.db.models.signals import post_migrate
from django.dispatch import receiver
from notify_catalog import NotifyCatalog
from models import NoticeType

@receiver(post_migrate)
def ensure_notice_kinds(sender, **kwargs):
    # Выполнять только для нужного приложения
    if sender.name != 'your_app_name':
        return

    current_ids = []
    for _, kind in NotifyCatalog._data.items():
        obj, _ = NoticeType.objects.update_or_create(
            name=kind.name,
            defaults={
                'description': kind.description,
                'is_active': True,
            },
        )
        current_ids.append(obj.id)

    # Деактивировать неиспользуемые уведомления
    NoticeType.objects.exclude(id__in=current_ids).update(is_active=False)

    # Обновить реестр в памяти
    NotifyCatalog._registry = {
        item.name: item
        for item in NoticeType.objects.filter(is_active=True)
    }

Поскольку post_migrate привязан к выполнению миграций, он не отработает при обычном runserver, если в этот запуск миграции не применяются. Чтобы инициализация была одинаковой, вызовите ту же логику из метода AppConfig.ready и одновременно зарегистрируйте обработчик post_migrate.

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

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

    def ready(self):
        # Запуск инициализации при старте сервера, если таблицы существуют
        if self._has_tables():
            self._seed_notice_kinds()

        # Обеспечить запуск и после миграций
        post_migrate.connect(self._after_migrate, sender=self)

    def _has_tables(self):
        from django.db import connection
        names = connection.introspection.table_names()
        return 'your_app_notificationtype' in names

    def _seed_notice_kinds(self):
        from .models import NoticeType
        from .notify_catalog import NotifyCatalog

        present_ids = []
        for _, kind in NotifyCatalog._data.items():
            obj, _ = NoticeType.objects.update_or_create(
                name=kind.name,
                defaults={
                    'description': kind.description,
                    'is_active': True,
                },
            )
            present_ids.append(obj.id)

        NoticeType.objects.exclude(id__in=present_ids).update(is_active=False)

        NotifyCatalog._registry = {
            item.name: item
            for item in NoticeType.objects.filter(is_active=True)
        }

    def _after_migrate(self, sender, **kwargs):
        if sender.name == self.name:
            self._seed_notice_kinds()

Зачем это важно

Сигналы, привязанные к соединениям с базой, срабатывают тогда, когда тестовый стенд ещё решает, к какой базе обращаться. Записи на уровне приложения в этот промежуточный момент вызывают трудноуловимые побочные эффекты и подрывают надежность тестов. Перенос синхронизации на post_migrate согласует начальное наполнение с готовностью схемы, а вызов из AppConfig.ready обеспечивает одинаковое поведение в разработке без зависимости от моментов установления соединения.

Итоги

Используйте connection_created для настроек сеанса базы данных, а не для заполнения или изменения таблиц приложения. Наполняйте или синхронизируйте типы уведомлений после миграций через post_migrate и вызывайте ту же процедуру из AppConfig.ready, когда таблицы уже есть. Так реестр остается источником истины, пропадают предупреждения под pytest с PostgreSQL, а подписки стабильно соответствуют корректному набору активных типов уведомлений в любой среде.