Assurer la stabilité des écouteurs de notification PostgreSQL de longue durée avec Psycopg3

Garantir des notifications fiables dans les écouteurs PostgreSQL de longue durée

Implémentation backend utilisant Python et psycopg3 pour gérer les connexions de base de données de longue durée

import psycopg
from psycopg import sql
import time
CONN_STR = "postgresql://user:password@localhost/dbname"
def listen_notifications():
    try:
        with psycopg.connect(CONN_STR, autocommit=True) as conn:
            listen_sql = sql.SQL("LISTEN {};").format(sql.Identifier("scheduler_test"))
            conn.execute(listen_sql)
            print("Listening for notifications...")
            gen = conn.notifies(timeout=5)
            for notification in gen:
                print("Received notification:", notification)
                perform_health_check(conn, listen_sql)
    except Exception as e:
        print("Error:", e)
def perform_health_check(conn, listen_sql):
    try:
        print("Performing health check...")
        conn.execute("SELECT 1")
        conn.execute(listen_sql)
    except Exception as e:
        print("Health check failed:", e)
if __name__ == "__main__":
    listen_notifications()

Approche alternative : utilisation de psycopg3 asynchrone pour une réactivité améliorée

Implémentation asynchrone utilisant asyncio et psycopg3 de Python

import asyncio
from psycopg import AsyncConnection, sql
CONN_STR = "postgresql://user:password@localhost/dbname"
async def listen_notifications():
    try:
        async with AsyncConnection.connect(CONN_STR, autocommit=True) as conn:
            listen_sql = sql.SQL("LISTEN {};").format(sql.Identifier("scheduler_test"))
            await conn.execute(listen_sql)
            print("Listening for notifications...")
            gen = conn.notifies(timeout=5)
            async for notification in gen:
                print("Received notification:", notification)
                await perform_health_check(conn, listen_sql)
    except Exception as e:
        print("Error:", e)
async def perform_health_check(conn, listen_sql):
    try:
        print("Performing health check...")
        await conn.execute("SELECT 1")
        await conn.execute(listen_sql)
    except Exception as e:
        print("Health check failed:", e)
if __name__ == "__main__":
    asyncio.run(listen_notifications())

Tests unitaires pour la robustesse

Tests unitaires Python pour la logique backend à l'aide de unittest

import unittest
from unittest.mock import patch, MagicMock
class TestNotificationListener(unittest.TestCase):
    @patch("psycopg.connect")
    def test_listen_notifications(self, mock_connect):
        mock_conn = MagicMock()
        mock_connect.return_value.__enter__.return_value = mock_conn
        mock_conn.notifies.return_value = iter(["test_notification"])
        listen_notifications()
        mock_conn.execute.assert_called_with("LISTEN scheduler_test;")
        mock_conn.notifies.assert_called_once()
if __name__ == "__main__":
    unittest.main()

FAQ sur la gestion des écouteurs de notification PostgreSQL

1. Quel est le but de conn.notifies() dans psycopg3 ?
2. conn.notifies() est utilisé pour récupérer les notifications envoyées par le serveur PostgreSQL, permettant la gestion des événements en temps réel dans les applications.
3. Peut LISTEN les commandes perdent des messages lors de la reconnexion ?
4. Non, PostgreSQL met en mémoire tampon les notifications afin que les messages ne soient pas perdus lors de la reconnexion. Cependant, une bonne manipulation du notifies Un générateur est nécessaire pour assurer un traitement transparent.
5. Pourquoi devrais-je utiliser autocommit=True?
6. Paramètre autocommit=True permet à la connexion d'appliquer immédiatement des commandes telles que LISTEN sans attendre un commit explicite, améliorant ainsi la réactivité.
7. Comment puis-je effectuer des contrôles de santé pendant une longue période notifies processus?
8. Vous pouvez exécuter périodiquement des requêtes légères comme SELECT 1 pour garantir que la connexion reste réactive.
9. Quelles sont les meilleures pratiques pour nettoyer les connexions aux bases de données ?
10. Utiliser un with L'instruction ou le gestionnaire de contexte de Python garantit que la connexion est correctement fermée, évitant ainsi les fuites de ressources.
11. Comment gérer les exceptions de délai d'attente dans conn.notifies()?
12. Envelopper conn.notifies() dans un bloc try-sauf pour intercepter les exceptions de délai d'attente et les gérer correctement, par exemple en vous connectant ou en réessayant.
13. Psychopg3 prend-il en charge les opérations asynchrones pour les notifications ?
14. Oui, psycopg3 propose une API asynchrone via AsyncConnection, ce qui est idéal pour les applications non bloquantes et évolutives.
15. Que se passe-t-il si je ne ferme pas le notifies générateur?
16. Ne pas fermer le générateur peut entraîner des fuites de mémoire ou des ressources bloquées, en particulier dans les processus de longue durée.
17. Les notifications peuvent-elles être manquées pendant un pg_sleep() opération?
18. Oui, les notifications générées pendant la période de veille peuvent être manquées si elles ne sont pas mises en mémoire tampon. C'est pourquoi une gestion appropriée des LISTEN les commandes sont cruciales.
19. Est-il sécuritaire de réutiliser la même connexion pour plusieurs notifications ?
20. Oui, tant que les contrôles de santé et les reconnexions appropriées sont gérés, la réutilisation de la même connexion est efficace et respectueuse des ressources.
21. Comment tester la fiabilité de mon système de notification ?
22. Écrivez des tests unitaires en utilisant des bibliothèques comme unittest.mock pour simuler les notifications et le comportement de la base de données sans recourir à un serveur en direct.

Assurer une écoute fiable des notifications

Le maintien de l’intégrité des connexions pour les processus de longue durée est essentiel pour des opérations ininterrompues. Avec les outils de psycopg3 comme conn.notifie(), les développeurs peuvent mettre en œuvre des systèmes de notification robustes. Des contrôles de santé réguliers permettent d’éviter les connexions qui ne répondent pas. Les exemples incluent la surveillance des systèmes d’inventaire pour les mises à jour en direct afin d’éviter les pannes.

La fermeture et la réouverture du générateur de notifications, combinées à des commandes SQL légères, garantissent à la fois performances et fiabilité. Ces techniques s'appliquent à divers cas d'utilisation, des mises à jour logistiques aux alertes financières. De telles stratégies contribuent à protéger les applications critiques contre les temps d'arrêt, garantissant ainsi une expérience utilisateur transparente. ⚡