Add EventQueueProtocol (#9)

Author: janbjorge

src/pgcachewatch/decorators.py

@@ -19,7 +19,7 @@ def outer(fn: Callable[P, Awaitable[T]]) -> Callable[P, Awaitable[T]]:
 
         async def inner(*args: P.args, **kwargs: P.kwargs) -> T:
             # If db-conn is down, disable cache.
-            if not strategy.pg_connection_healthy():
+            if not strategy.connection_healthy():
                 logging.critical("Database connection is closed, caching disabled.")
                 return await fn(*args, **kwargs)
 

src/pgcachewatch/listeners.py

@@ -2,6 +2,7 @@
 import datetime
 import json
 import logging
+from typing import Protocol
 
 import asyncpg
 
@@ -15,6 +16,44 @@ def _critical_termination_listener(*_: object, **__: object) -> None:
     logging.critical("Connection is closed / terminated.")
 
 
+class EventQueueProtocol(Protocol):
+    """
+    Protocol for an event queue interface.
+
+    Specifies the required methods for an event queue to check the connection health
+    and to retrieve events without waiting. Implementing classes must provide concrete
+    implementations of these methods to ensure compatibility with the event handling
+    system.
+    """
+
+    def connection_healthy(self) -> bool:
+        """
+        Checks if the connection is healthy.
+
+        This method should return True if the connection to the underlying service
+        (e.g., database, message broker) is active and healthy, False otherwise.
+
+        Returns:
+            bool: True if the connection is healthy, False otherwise.
+        """
+        raise NotImplementedError
+
+    def get_nowait(self) -> models.Event:
+        """
+        Retrieves an event from the queue without waiting.
+
+        Attempts to immediately retrieve an event from the queue. If no event is
+        available, this method should raise an appropriate exception (e.g., QueueEmpty).
+
+        Returns:
+            models.Event: The event retrieved from the queue.
+
+        Raises:
+            QueueEmpty: If no event is available in the queue to retrieve.
+        """
+        raise NotImplementedError
+
+
 class PGEventQueue(asyncio.Queue[models.Event]):
     """
     A PostgreSQL event queue that listens to a specified
@@ -34,7 +73,7 @@ def __init__(
     async def connect(
         self,
         connection: asyncpg.Connection,
-        channel: models.PGChannel,
+        channel: models.PGChannel = models.PGChannel("ch_pgcachewatch_table_change"),
     ) -> None:
         """
         Asynchronously connects the PGEventQueue to a specified
@@ -104,5 +143,5 @@ def parse_and_put(
             except Exception:
                 logging.exception("Unable to queue `%s`.", parsed)
 
-    def pg_connection_healthy(self) -> bool:
+    def connection_healthy(self) -> bool:
         return bool(self._pg_connection and not self._pg_connection.is_closed())

src/pgcachewatch/strategies.py

@@ -13,7 +13,7 @@ class Strategy(typing.Protocol):
     def clear(self) -> bool:
         raise NotImplementedError
 
-    def pg_connection_healthy(self) -> bool:
+    def connection_healthy(self) -> bool:
         raise NotImplementedError
 
 
@@ -24,22 +24,22 @@ class Gready(Strategy):
 
     def __init__(
         self,
-        listener: listeners.PGEventQueue,
-        deadline: models.DeadlineSetting = models.DeadlineSetting(),
+        listener: listeners.EventQueueProtocol,
+        settings: models.DeadlineSetting = models.DeadlineSetting(),
         predicate: typing.Callable[[models.Event], bool] = bool,
     ) -> None:
         super().__init__()
         self._listener = listener
         self._predicate = predicate
-        self._deadline = deadline
+        self._settings = settings
 
-    def pg_connection_healthy(self) -> bool:
-        return self._listener.pg_connection_healthy()
+    def connection_healthy(self) -> bool:
+        return self._listener.connection_healthy()
 
     def clear(self) -> bool:
         for current in utils.pick_until_deadline(
             self._listener,
-            settings=self._deadline,
+            settings=self._settings,
         ):
             if self._predicate(current):
                 return True
@@ -54,23 +54,23 @@ class Windowed(Strategy):
 
     def __init__(
         self,
-        listener: listeners.PGEventQueue,
+        listener: listeners.EventQueueProtocol,
         window: list[models.OPERATIONS],
-        deadline: models.DeadlineSetting = models.DeadlineSetting(),
+        settings: models.DeadlineSetting = models.DeadlineSetting(),
     ) -> None:
         super().__init__()
         self._listener = listener
         self._window = window
-        self._deadline = deadline
+        self._settings = settings
         self._events = collections.deque[models.OPERATIONS](maxlen=len(self._window))
 
-    def pg_connection_healthy(self) -> bool:
-        return self._listener.pg_connection_healthy()
+    def connection_healthy(self) -> bool:
+        return self._listener.connection_healthy()
 
     def clear(self) -> bool:
         for current in utils.pick_until_deadline(
             self._listener,
-            settings=self._deadline,
+            settings=self._settings,
         ):
             self._events.append(current.operation)
             if len(self._window) == len(self._events) and all(
@@ -87,23 +87,23 @@ class Timed(Strategy):
 
     def __init__(
         self,
-        listener: listeners.PGEventQueue,
+        listener: listeners.EventQueueProtocol,
         timedelta: datetime.timedelta,
-        deadline: models.DeadlineSetting = models.DeadlineSetting(),
+        settings: models.DeadlineSetting = models.DeadlineSetting(),
     ) -> None:
         super().__init__()
         self._listener = listener
         self._timedelta = timedelta
-        self._deadline = deadline
+        self._settings = settings
         self._previous = datetime.datetime.now(tz=datetime.timezone.utc)
 
-    def pg_connection_healthy(self) -> bool:
-        return self._listener.pg_connection_healthy()
+    def connection_healthy(self) -> bool:
+        return self._listener.connection_healthy()
 
     def clear(self) -> bool:
         for current in utils.pick_until_deadline(
             queue=self._listener,
-            settings=self._deadline,
+            settings=self._settings,
         ):
             if current.sent_at - self._previous > self._timedelta:
                 self._previous = current.sent_at

src/pgcachewatch/utils.py

@@ -34,7 +34,7 @@ async def emit_event(
 
 
 def pick_until_deadline(
-    queue: listeners.PGEventQueue,
+    queue: listeners.EventQueueProtocol,
     settings: models.DeadlineSetting,
 ) -> typing.Iterator[models.Event]:
     """