Garantindo estabilidade em ouvintes de notificação PostgreSQL de longa duração com Psycopg3

Garantindo notificações confiáveis ​​em ouvintes PostgreSQL de longa duração

Implementação de back-end usando Python e psycopg3 para lidar com conexões de banco de dados de longa duração

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

Abordagem alternativa: usando psycopg3 assíncrono para maior capacidade de resposta

Implementação assíncrona usando asyncio e psycopg3 do 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())

Teste de unidade para robustez

Testes de unidade Python para lógica de back-end 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()

Perguntas frequentes sobre como gerenciar ouvintes de notificação do PostgreSQL

1. Qual é o propósito conn.notifies() em psicopg3?
2. conn.notifies() é usado para recuperar notificações enviadas pelo servidor PostgreSQL, permitindo o tratamento de eventos em tempo real nas aplicações.
3. Pode LISTEN comandos perdem mensagens durante a reconexão?
4. Não, o PostgreSQL armazena notificações em buffer, para que as mensagens não sejam perdidas durante a reconexão. Contudo, o manejo adequado do notifies gerador é necessário para garantir um processamento contínuo.
5. Por que devo usar autocommit=True?
6. Contexto autocommit=True permite que a conexão aplique imediatamente comandos como LISTEN sem esperar por um commit explícito, melhorando a capacidade de resposta.
7. Como posso realizar verificações de integridade durante uma operação de longa duração notifies processo?
8. Você pode executar periodicamente consultas leves como SELECT 1 para garantir que a conexão permaneça responsiva.
9. Quais são as práticas recomendadas para limpar conexões de banco de dados?
10. Usando um with instrução ou gerenciador de contexto do Python garante que a conexão seja fechada corretamente, evitando vazamentos de recursos.
11. Como lidar com exceções de tempo limite em conn.notifies()?
12. Enrolar conn.notifies() em um bloco try-except para capturar exceções de tempo limite e tratá-las normalmente, como registrando ou tentando novamente.
13. O psycopg3 oferece suporte a operações assíncronas para notificações?
14. Sim, o psycopg3 oferece uma API assíncrona via AsyncConnection, que é ideal para aplicativos escalonáveis ​​e sem bloqueio.
15. O que acontece se eu não fechar o notifies gerador?
16. Não fechar o gerador pode resultar em vazamentos de memória ou suspensão de recursos, especialmente em processos de longa execução.
17. As notificações podem ser perdidas durante um pg_sleep() operação?
18. Sim, as notificações geradas durante o período de suspensão podem ser perdidas se não forem armazenadas em buffer, e é por isso que o manuseio adequado de LISTEN comandos é crucial.
19. É seguro reutilizar a mesma conexão para múltiplas notificações?
20. Sim, desde que as verificações de integridade e as reconexões adequadas sejam gerenciadas, a reutilização da mesma conexão é eficiente e econômica.
21. Como posso testar a confiabilidade do meu sistema de notificação?
22. Escreva testes unitários usando bibliotecas como unittest.mock para simular notificações e comportamento do banco de dados sem depender de um servidor ativo.

Garantindo uma escuta confiável de notificações

Manter a integridade da conexão para processos de longa execução é essencial para operações ininterruptas. Com as ferramentas do psycopg3 como conn.notifica(), os desenvolvedores podem implementar sistemas de notificação robustos. Verificações regulares de integridade ajudam a evitar conexões que não respondem. Os exemplos incluem o monitoramento de sistemas de inventário para atualizações em tempo real para evitar interrupções.

Fechar e reabrir o gerador de notificações, combinado com comandos SQL leves, garante desempenho e confiabilidade. Essas técnicas se aplicam a vários casos de uso, desde atualizações logísticas até alertas financeiros. Essas estratégias ajudam a proteger aplicações críticas contra tempo de inatividade, garantindo uma experiência de usuário perfeita. ⚡