Garantire la stabilità negli ascoltatori di notifiche PostgreSQL di lunga durata con Psycopg3

Garantire notifiche affidabili nei listener PostgreSQL di lunga durata

Implementazione del backend utilizzando Python e psycopg3 per gestire connessioni al database di lunga durata

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()

Approccio alternativo: utilizzo di psycopg3 asincrono per una maggiore reattività

Implementazione asincrona utilizzando asyncio e psycopg3 di 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())

Test unitari per robustezza

Test unitari Python per la logica di backend utilizzando 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()

Domande frequenti sulla gestione dei listener di notifiche PostgreSQL

1. Qual è lo scopo di conn.notifies() in psycopg3?
2. conn.notifies() viene utilizzato per recuperare le notifiche inviate dal server PostgreSQL, consentendo la gestione degli eventi in tempo reale nelle applicazioni.
3. Potere LISTEN i comandi perdono i messaggi durante la riconnessione?
4. No, PostgreSQL memorizza nel buffer le notifiche, quindi i messaggi non vengono persi durante la riconnessione. Tuttavia, la corretta gestione del notifies è necessario un generatore per garantire un'elaborazione senza interruzioni.
5. Perché dovrei usare autocommit=True?
6. Collocamento autocommit=True consente alla connessione di applicare immediatamente comandi come LISTEN senza attendere un commit esplicito, migliorando la reattività.
7. Come posso eseguire controlli sanitari durante un lungo periodo notifies processo?
8. Puoi eseguire periodicamente query leggere come SELECT 1 per garantire che la connessione rimanga reattiva.
9. Quali sono le migliori pratiche per pulire le connessioni al database?
10. Utilizzando a with o il gestore del contesto di Python garantisce che la connessione sia chiusa correttamente, evitando perdite di risorse.
11. Come posso gestire le eccezioni di timeout in conn.notifies()?
12. Avvolgere conn.notifies() in un blocco try-eccetto per rilevare le eccezioni di timeout e gestirle con garbo, ad esempio effettuando la registrazione o riprovando.
13. psycopg3 supporta operazioni asincrone per le notifiche?
14. Sì, psycopg3 offre un'API asincrona tramite AsyncConnection, ideale per applicazioni scalabili e non bloccanti.
15. Cosa succede se non chiudo il file notifies generatore?
16. La mancata chiusura del generatore potrebbe causare perdite di memoria o risorse bloccate, soprattutto nei processi di lunga durata.
17. È possibile perdere le notifiche durante a pg_sleep() operazione?
18. Sì, le notifiche generate durante il periodo di sospensione potrebbero andare perse se non memorizzate nel buffer, motivo per cui vengono gestite correttamente LISTEN i comandi sono cruciali.
19. È sicuro riutilizzare la stessa connessione per più notifiche?
20. Sì, purché vengano gestiti i controlli di integrità e le riconnessioni adeguate, riutilizzare la stessa connessione è efficiente e rispettoso delle risorse.
21. Come posso testare l'affidabilità del mio sistema di notifica?
22. Scrivi test unitari utilizzando librerie come unittest.mock per simulare notifiche e comportamento del database senza fare affidamento su un server live.

Garantire un ascolto affidabile delle notifiche

Mantenere l'integrità della connessione per i processi di lunga durata è essenziale per operazioni ininterrotte. Con gli strumenti di psycopg3 come conn.notifica(), gli sviluppatori possono implementare robusti sistemi di notifica. Controlli regolari dello stato aiutano a evitare connessioni che non rispondono. Gli esempi includono il monitoraggio dei sistemi di inventario per aggiornamenti in tempo reale per prevenire interruzioni.

La chiusura e la riapertura del generatore di notifiche, combinata con comandi SQL leggeri, garantisce prestazioni e affidabilità. Queste tecniche si applicano a vari casi d'uso, dagli aggiornamenti logistici agli avvisi finanziari. Tali strategie aiutano a salvaguardare le applicazioni critiche dai tempi di inattività, garantendo un'esperienza utente senza interruzioni. ⚡