feat: automaticall call the update handler if it is provided

Author: trbtm

postgresql_watcher/watcher.py

@@ -69,7 +69,7 @@ def __init__(
         self.child_conn: Connection | None = None
         self.subscription_process: Process | None = None
         self._create_subscription_process(start_listening)
-        self.update_callback: Optional[Callable] = None
+        self.update_callback: Optional[Callable[[None], None]] = None
 
     def __del__(self) -> None:
         self._cleanup_connections_and_processes()
@@ -129,7 +129,7 @@ def _cleanup_connections_and_processes(self) -> None:
             self.subscription_process.terminate()
             self.subscription_process = None
 
-    def set_update_callback(self, update_handler: Callable):
+    def set_update_callback(self, update_handler: Optional[Callable[[None], None]]):
         """
         Set the handler called, when the Watcher detects an update.
         Recommendation: `casbin_enforcer.adapter.load_policy`
@@ -164,7 +164,10 @@ def should_reload(self) -> bool:
         try:
             if self.parent_conn.poll():
                 message = int(self.parent_conn.recv())
-                return message == _ChannelSubscriptionMessage.RECEIVED_UPDATE
+                received_update = message == _ChannelSubscriptionMessage.RECEIVED_UPDATE
+                if received_update and self.update_callback is not None:
+                    self.update_callback()
+                return received_update
         except EOFError:
             self.logger.warning(
                 "Child casbin-watcher subscribe process has stopped, "

tests/test_postgresql_watcher.py

@@ -1,5 +1,6 @@
 import sys
 import unittest
+from unittest.mock import MagicMock
 from multiprocessing.connection import Pipe
 from time import sleep
 import logging
@@ -80,6 +81,26 @@ def test_no_update_mutiple_pg_watcher(self):
         sleep(CASBIN_CHANNEL_SELECT_TIMEOUT * 2)
         for watcher in other_watchers:
             self.assertFalse(watcher.should_reload())
+        self.assertFalse(main_watcher.should_reload())
+
+    def test_update_handler_called(self):
+        channel_name = "test_update_handler_called"
+        main_watcher = get_watcher(channel_name)
+        handler = MagicMock()
+        main_watcher.set_update_callback(handler)
+        main_watcher.update()
+        sleep(CASBIN_CHANNEL_SELECT_TIMEOUT * 2)
+        self.assertTrue(main_watcher.should_reload())
+        self.assertTrue(handler.call_count == 1)
+
+    def test_update_handler_not_called(self):
+        channel_name = "test_update_handler_not_called"
+        main_watcher = get_watcher(channel_name)
+        handler = MagicMock()
+        main_watcher.set_update_callback(handler)
+        sleep(CASBIN_CHANNEL_SELECT_TIMEOUT * 2)
+        self.assertFalse(main_watcher.should_reload())
+        self.assertTrue(handler.call_count == 0)
 
 
 if __name__ == "__main__":