Garantizar la estabilidad en los oyentes de notificaciones de PostgreSQL de larga duración con Psycopg3

Garantizar notificaciones confiables en oyentes de PostgreSQL de larga duración

Implementación de backend usando Python y psycopg3 para manejar conexiones de bases de datos de larga duración

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

Enfoque alternativo: uso de psycopg3 asincrónico para mejorar la capacidad de respuesta

Implementación asincrónica usando asyncio y 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())

Pruebas unitarias de robustez

Pruebas unitarias de Python para lógica de backend usando 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()

Preguntas frecuentes sobre la gestión de escuchas de notificaciones de PostgreSQL

1. ¿Cuál es el propósito de conn.notifies() en psicopg3?
2. conn.notifies() se utiliza para recuperar notificaciones enviadas por el servidor PostgreSQL, lo que permite el manejo de eventos en tiempo real en las aplicaciones.
3. Poder LISTEN ¿Los comandos pierden mensajes durante la reconexión?
4. No, PostgreSQL almacena las notificaciones, por lo que los mensajes no se pierden durante la reconexión. Sin embargo, el manejo adecuado de la notifies Se requiere un generador para garantizar un procesamiento perfecto.
5. ¿Por qué debería usar autocommit=True?
6. Configuración autocommit=True permite que la conexión aplique inmediatamente comandos como LISTEN sin esperar un compromiso explícito, mejorando la capacidad de respuesta.
7. ¿Cómo puedo realizar controles de estado durante un proceso de larga duración? notifies ¿proceso?
8. Puede ejecutar periódicamente consultas ligeras como SELECT 1 para garantizar que la conexión siga respondiendo.
9. ¿Cuáles son las mejores prácticas para limpiar las conexiones de bases de datos?
10. Usando un with La declaración o el administrador de contexto de Python garantiza que la conexión se cierre correctamente, evitando fugas de recursos.
11. ¿Cómo manejo las excepciones de tiempo de espera en conn.notifies()?
12. Envoltura conn.notifies() en un bloque try-except para detectar excepciones de tiempo de espera y manejarlas correctamente, como iniciando sesión o reintentando.
13. ¿Admite psycopg3 operaciones asincrónicas para notificaciones?
14. Sí, psycopg3 ofrece una API asíncrona a través de AsyncConnection, que es ideal para aplicaciones escalables y sin bloqueo.
15. ¿Qué pasa si no cierro el notifies ¿generador?
16. No cerrar el generador puede provocar pérdidas de memoria o recursos bloqueados, especialmente en procesos de larga duración.
17. ¿Se pueden perder las notificaciones durante una pg_sleep() ¿operación?
18. Sí, las notificaciones generadas durante el período de suspensión pueden perderse si no se almacenan en el buffer, razón por la cual se maneja adecuadamente LISTEN comandos es crucial.
19. ¿Es seguro reutilizar la misma conexión para múltiples notificaciones?
20. Sí, siempre que se gestionen los controles de estado y las reconexiones adecuadas, reutilizar la misma conexión es eficiente y ahorra recursos.
21. ¿Cómo puedo probar la confiabilidad de mi sistema de notificación?
22. Escriba pruebas unitarias usando bibliotecas como unittest.mock para simular notificaciones y comportamiento de la base de datos sin depender de un servidor en vivo.

Garantizar una escucha de notificaciones confiable

Mantener el estado de la conexión para procesos de larga duración es esencial para operaciones ininterrumpidas. Con las herramientas de psycopg3 como conexión.notifica(), los desarrolladores pueden implementar sistemas de notificación sólidos. Los controles de estado periódicos ayudan a evitar conexiones que no responden. Los ejemplos incluyen el monitoreo de los sistemas de inventario para obtener actualizaciones en vivo para evitar interrupciones.

Cerrar y volver a abrir el generador de notificaciones, combinado con comandos SQL ligeros, garantiza rendimiento y confiabilidad. Estas técnicas se aplican a diversos casos de uso, desde actualizaciones logísticas hasta alertas financieras. Estas estrategias ayudan a proteger las aplicaciones críticas contra el tiempo de inactividad, garantizando una experiencia de usuario perfecta. ⚡