Extract shared connection lifecycle into ConnectionAwareAdapter.

Move duplicated context-manager, health-check cache, and connection guards from Redis and Minio adapters into an internal base class.

Co-authored-by: Cursor <[email protected]>

python_repositories/adapters/connection_aware_adapter.py

@@ -0,0 +1,80 @@
+"""Shared connection lifecycle behavior for repository adapters."""
+
+from __future__ import annotations
+
+import time
+from abc import abstractmethod
+from typing import Self
+
+import structlog
+
+from python_repositories.interfaces import (
+    ConnectionAwareInterface,
+    ContextAwareInterface,
+)
+
+
+class ConnectionAwareAdapter(ConnectionAwareInterface, ContextAwareInterface):
+    """Base adapter with context-manager and TTL-cached connection health checks."""
+
+    health_check_ttl_seconds: float = 1.0
+    connection_name: str = ""
+
+    def __init__(self) -> None:
+        self.logger = structlog.get_logger(self.__class__.__name__)
+        self._health_check_at: float | None = None
+        self._health_check_ok: bool = False
+
+    def __enter__(self) -> Self:
+        """Enter the context."""
+        self.connect()
+        return self
+
+    def __exit__(
+        self, exc_type: type | None, exc_val: object | None, exc_tb: object | None
+    ) -> None:
+        """Exit the context."""
+        ctx_info = {"exc_type": exc_type, "exc_val": exc_val, "exc_tb": exc_tb}
+        if any(
+            (
+                exc_type is not None,
+                exc_val is not None,
+                exc_tb is not None,
+            ),
+        ):
+            self.logger.error("Error while exiting context", **ctx_info)
+        self.disconnect()
+
+    def _invalidate_health_cache(self) -> None:
+        self._health_check_at = None
+        self._health_check_ok = False
+
+    @abstractmethod
+    def _is_client_ready(self) -> bool:
+        """Return True when internal state is sufficient for a probe."""
+
+    @abstractmethod
+    def _probe_connection(self) -> bool:
+        """Backend-specific liveness check; called only when client is ready."""
+
+    def is_connected(self) -> bool:
+        """Check if connected to the backend."""
+        if not self._is_client_ready():
+            return False
+        now = time.monotonic()
+        if self._health_check_at is not None:
+            seconds_since_last_health_check = now - self._health_check_at
+            cache_is_fresh = (
+                seconds_since_last_health_check < self.health_check_ttl_seconds
+            )
+            if cache_is_fresh:
+                return self._health_check_ok
+        result = self._probe_connection()
+        self._health_check_at = now
+        self._health_check_ok = result
+        self.logger.debug("Connection status", connected=result)
+        return result
+
+    def _require_connected(self) -> None:
+        if not self.is_connected():
+            raise ConnectionError(f"Not connected to {self.connection_name}")