Psycopg3 এর সাথে দীর্ঘ-চলমান PostgreSQL বিজ্ঞপ্তি শ্রোতাদের মধ্যে স্থিতিশীলতা নিশ্চিত করা

দীর্ঘ-চলমান PostgreSQL শ্রোতাদের মধ্যে নির্ভরযোগ্য বিজ্ঞপ্তি নিশ্চিত করা

দীর্ঘস্থায়ী ডাটাবেস সংযোগগুলি পরিচালনা করতে পাইথন এবং সাইকোপজি 3 ব্যবহার করে ব্যাকএন্ড বাস্তবায়ন

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

বিকল্প পদ্ধতি: বর্ধিত প্রতিক্রিয়াশীলতার জন্য অ্যাসিঙ্ক্রোনাস সাইকোপজি3 ব্যবহার করা

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

দৃঢ়তার জন্য ইউনিট পরীক্ষা

ইউনিটটেস্ট ব্যবহার করে ব্যাকএন্ড লজিকের জন্য পাইথন ইউনিট পরীক্ষা করে

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 স্টেটমেন্ট বা পাইথনের কনটেক্সট ম্যানেজার নিশ্চিত করে যে সংযোগটি সঠিকভাবে বন্ধ করা হয়েছে, রিসোর্স লিক এড়ানো।
11. কিভাবে আমি টাইমআউট ব্যতিক্রমগুলি পরিচালনা করব conn.notifies()?
12. মোড়ানো conn.notifies() টাইমআউট ব্যতিক্রমগুলি ধরার জন্য ব্লক ব্যতীত চেষ্টা করুন এবং সেগুলিকে সুন্দরভাবে পরিচালনা করুন, যেমন লগিং বা পুনরায় চেষ্টা করে।
13. psycopg3 কি বিজ্ঞপ্তির জন্য অ্যাসিঙ্ক্রোনাস অপারেশন সমর্থন করে?
14. হ্যাঁ, psycopg3 এর মাধ্যমে একটি অ্যাসিঙ্ক্রোনাস API অফার করে AsyncConnection, যা নন-ব্লকিং, মাপযোগ্য অ্যাপ্লিকেশনের জন্য আদর্শ।
15. আমি বন্ধ না করলে কি হবে notifies জেনারেটর?
16. জেনারেটর বন্ধ করতে ব্যর্থ হলে মেমরি লিক হতে পারে বা রিসোর্স ঝুলে যেতে পারে, বিশেষ করে দীর্ঘ-চলমান প্রক্রিয়ায়।
17. একটি সময় বিজ্ঞপ্তি মিস করা যেতে পারে pg_sleep() অপারেশন?
18. হ্যাঁ, ঘুমের সময় উত্পন্ন বিজ্ঞপ্তিগুলি বাফার না হলে মিস হতে পারে, এই কারণেই এর সঠিক পরিচালনা LISTEN আদেশ অত্যন্ত গুরুত্বপূর্ণ।
19. একাধিক বিজ্ঞপ্তির জন্য একই সংযোগ পুনরায় ব্যবহার করা কি নিরাপদ?
20. হ্যাঁ, যতক্ষণ পর্যন্ত স্বাস্থ্য পরীক্ষা এবং যথাযথ পুনঃসংযোগ পরিচালনা করা হয়, একই সংযোগ পুনঃব্যবহার করা দক্ষ এবং সম্পদ-বান্ধব।
21. আমি কিভাবে আমার বিজ্ঞপ্তি সিস্টেমের নির্ভরযোগ্যতা পরীক্ষা করতে পারি?
22. লাইব্রেরি ব্যবহার করে ইউনিট পরীক্ষা লিখুন unittest.mock একটি লাইভ সার্ভারের উপর নির্ভর না করে বিজ্ঞপ্তি এবং ডাটাবেস আচরণ অনুকরণ করতে।

নির্ভরযোগ্য বিজ্ঞপ্তি শোনা নিশ্চিত করা

নিরবচ্ছিন্ন ক্রিয়াকলাপের জন্য দীর্ঘ-চলমান প্রক্রিয়াগুলির জন্য সংযোগের স্বাস্থ্য বজায় রাখা অপরিহার্য। psycopg3 এর মত টুল সহ conn.notifies(), বিকাশকারীরা শক্তিশালী বিজ্ঞপ্তি সিস্টেম বাস্তবায়ন করতে পারে। নিয়মিত স্বাস্থ্য পরীক্ষা অপ্রতিক্রিয়াশীল সংযোগ এড়াতে সাহায্য করে। উদাহরণগুলির মধ্যে রয়েছে বিভ্রাট প্রতিরোধ করার জন্য লাইভ আপডেটের জন্য ইনভেন্টরি সিস্টেমগুলি পর্যবেক্ষণ করা।

লাইটওয়েট এসকিউএল কমান্ডের সাথে মিলিত নোটিফাই জেনারেটর বন্ধ করা এবং পুনরায় খোলার ফলে কর্মক্ষমতা এবং নির্ভরযোগ্যতা উভয়ই নিশ্চিত হয়। এই কৌশলগুলি বিভিন্ন ব্যবহারের ক্ষেত্রে প্রযোজ্য, লজিস্টিক আপডেট থেকে শুরু করে আর্থিক সতর্কতা পর্যন্ত। এই ধরনের কৌশলগুলি একটি নিরবচ্ছিন্ন ব্যবহারকারীর অভিজ্ঞতা নিশ্চিত করে, ডাউনটাইমের বিরুদ্ধে সমালোচনামূলক অ্যাপ্লিকেশনগুলিকে রক্ষা করতে সহায়তা করে। ⚡