connection_created से बचें: Django में post_migrate और AppConfig.ready से नोटिफिकेशन सिंक

Django में नोटिफिकेशन प्रकारों का सिंक सीखें: connection_created के बजाय post_migrate और AppConfig.ready अपनाएँ—pytest तथा PostgreSQL चेतावनियाँ खत्म.

कोड और डेटाबेस के बीच नोटिफिकेशन प्रकारों की एक मानक सूची को सिंक में रखना पहली नज़र में आसान लगता है, लेकिन जैसे ही टेस्ट आते हैं, तस्वीर बदल जाती है. लोकल रन में एक साधारण connection‑level हुक काम करता दिखता है, मगर PostgreSQL के साथ pytest पर वह शोर भरी चेतावनियाँ उछाल देता है. नीचे चरण-दर-चरण समझाया है कि दिक्कत कहाँ होती है और runserver तथा टेस्ट—दोनों वातावरणों में इसे भरोसेमंद कैसे बनाया जाए.

सेटअप: DB कनेक्शन पर नोटिफिकेशन प्रकारों का सिंक

लक्ष्य यह है कि समर्थित नोटिफिकेशन प्रकारों का रजिस्ट्रि डेटाबेस में रखा जाए, उनकी मेटाडेटा जानकारी बनाए रखी जाए, और ग्राहकों के साथ many‑to‑many मैपिंग हो. हर नोटिफिकेशन की लॉजिक अपनी कक्षा (class) में रहती है, और एक सिंगलटन जैसा कंटेनर उनका मेटाडेटा एकत्र करता है. पहली कोशिश थी कि जैसे ही कनेक्शन खुले, 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 लिखने (writes) चलाते हैं, तो वे उस पल चल जाते हैं जब ऐप की डेटाबेस सामान्य कामकाज के लिए पूरी तरह तैयार भी नहीं होती. यही टाइमिंग मिसमैच RuntimeWarning के रूप में दिखता है कि टेस्ट सेटअप के दौरान Django पहले PostgreSQL डेटाबेस पर फॉलबैक कर रहा है. यानी यह हुक लो‑लेवल कनेक्शन ट्वीक के लिए ठीक है, पर टेस्टिंग इंफ्रास्ट्रक्चर के बूटस्ट्रैप के बीच एप्लिकेशन टेबल्स को सीड/म्यूटेट करने में यह नाज़ुक पड़ जाता है.

स्थिर समाधान: seeding को post_migrate तक टालें, और runserver को AppConfig.ready से कवर करें

सिंक्रनाइज़ेशन को लाइफ़साइकल के बाद के चरण तक टाल देने से टेस्ट डेटाबेस बनने के साथ होने वाली रेस खत्म हो जाती है. post_migrate सिग्नल माइग्रेशन लागू होने के बाद चलता है—यही वह सही समय है जब मेटाडेटा पंक्तियाँ upsert करनी चाहिए. सामान्य सर्वर स्टार्टअप पर भी वही इनिशियलाइज़ेशन 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()

यह क्यों मायने रखता है

डेटाबेस कनेक्शन से जुड़े सिग्नल उस समय चलते हैं जब टेस्टिंग हार्नेस अभी यह तय ही कर रहा होता है कि किस डेटाबेस पर काम करना है. उस बीच एप्लिकेशन‑स्तर के writes चलाने से ऐसे दुष्प्रभाव पैदा होते हैं जिन्हें पकड़ना मुश्किल होता है, और टेस्ट की विश्वसनीयता घटती है. सिंक्रनाइज़ेशन को post_migrate पर ले जाने से डेटा सीडिंग स्कीमा‑레डिनेस के साथ संरेखित हो जाती है, और AppConfig.ready के जरिए runserver को कवर करने से डेवलपमेंट में व्यवहार स्थिर रहता है—कनेक्शन टाइमिंग पर निर्भरता के बिना.

संक्षेप

connection_created का उपयोग डेटाबेस‑सेशन स्तर के समायोजनों के लिए करें, न कि एप्लिकेशन टेबल्स को भरने या बदलने के लिए. माइग्रेशन के बाद post_migrate से नोटिफिकेशन प्रकारों को सीड/सिंक करें, और जब टेबल पहले से मौजूद हों तो वही रूटीन AppConfig.ready से कॉल करें. इससे आपकी रजिस्ट्री विश्वसनीय बनी रहती है, PostgreSQL के साथ pytest में चेतावनियाँ नहीं आतीं, और हर वातावरण में सब्स्क्रिप्शन सही‑सही सक्रिय नोटिफिकेशन प्रकारों से मैप होते हैं.