Забезпечення стабільності довготривалих прослуховувачів сповіщень PostgreSQL за допомогою Psycopg3

Забезпечення надійних сповіщень у довготривалих слухачах PostgreSQL

Реалізація бекенда з використанням Python і psycopg3 для обробки тривалих підключень до бази даних

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

Альтернативний підхід: використання асинхронного psycopg3 для покращення реагування

Асинхронна реалізація з використанням Python asyncio та psycopg3

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

Модульне тестування на надійність

Модульні тести Python для серверної логіки за допомогою 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()

Поширені запитання щодо керування слухачами сповіщень PostgreSQL

1. Яка мета conn.notifies() в psycopg3?
2. conn.notifies() використовується для отримання сповіщень, надісланих сервером PostgreSQL, уможливлюючи обробку подій у режимі реального часу в програмах.
3. може LISTEN команди втрачають повідомлення під час повторного підключення?
4. Ні, PostgreSQL буферизує сповіщення, тому повідомлення не втрачаються під час повторного підключення. Однак належне поводження з notifies генератор необхідний для забезпечення безперебійної обробки.
5. Чому я повинен використовувати autocommit=True?
6. Налаштування autocommit=True дозволяє підключенню негайно застосовувати такі команди, як LISTEN не чекаючи явної фіксації, покращуючи реакцію.
7. Як я можу виконати перевірку стану здоров’я під час тривалої роботи notifies процес?
8. Ви можете періодично виконувати легкі запити, наприклад SELECT 1 щоб з’єднання залишалося чуйним.
9. Які найкращі методи очищення підключень до бази даних?
10. Використовуючи a with або менеджер контексту Python гарантує, що з’єднання закрито належним чином, уникаючи витоків ресурсів.
11. Як обробляти винятки часу очікування в conn.notifies()?
12. Обгортка conn.notifies() у блоці try-except, щоб перехопити винятки часу очікування та акуратно їх обробляти, наприклад шляхом реєстрації або повторної спроби.
13. Чи підтримує psycopg3 асинхронні операції для сповіщень?
14. Так, psycopg3 пропонує асинхронний API через AsyncConnection, що ідеально підходить для неблокуючих масштабованих програм.
15. Що станеться, якщо я не закрию notifies генератор?
16. Якщо генератор не закрити, це може призвести до витоків пам’яті або зависання ресурсів, особливо в довготривалих процесах.
17. Чи можна пропустити сповіщення під час a pg_sleep() операція?
18. Так, сповіщення, створені під час періоду сну, можуть бути пропущені, якщо їх не буферизувати, тому належне поводження з LISTEN команд має вирішальне значення.
19. Чи безпечно повторно використовувати те саме з’єднання для кількох сповіщень?
20. Так, за умови керування перевірками працездатності та належним повторним підключенням повторне використання того самого з’єднання є ефективним і ресурсозберігаючим.
21. Як я можу перевірити надійність моєї системи сповіщень?
22. Напишіть модульні тести, використовуючи такі бібліотеки, як unittest.mock для імітації сповіщень і поведінки бази даних, не покладаючись на живий сервер.

Забезпечення надійного прослуховування сповіщень

Підтримка працездатності підключення для тривалих процесів є важливою для безперебійної роботи. З такими інструментами psycopg3, як conn.notifyes(), розробники можуть впроваджувати надійні системи сповіщень. Регулярні перевірки працездатності допомагають уникнути з’єднань, які не реагують. Приклади включають моніторинг систем інвентаризації для оперативних оновлень для запобігання збоям.

Закриття та повторне відкриття генератора сповіщень у поєднанні зі спрощеними командами SQL забезпечує продуктивність і надійність. Ці методи застосовуються до різних випадків використання, від оновлень логістики до фінансових сповіщень. Такі стратегії допомагають захистити критично важливі програми від простою, забезпечуючи безперебійну роботу користувача. ⚡