From 7b950e0d93c95e426b3e0462bb83e021d7cde782 Mon Sep 17 00:00:00 2001 From: Brian Bjarke Jensen Date: Sun, 5 Jul 2026 16:09:25 +0200 Subject: [PATCH] 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 --- .../adapters/connection_aware_adapter.py | 80 ++++++++++++++++++ python_repositories/adapters/minio_adapter.py | 83 +++---------------- python_repositories/adapters/redis_adapter.py | 83 +++---------------- tests/unit/connection_health_test.py | 16 ++-- 4 files changed, 114 insertions(+), 148 deletions(-) create mode 100644 python_repositories/adapters/connection_aware_adapter.py diff --git a/python_repositories/adapters/connection_aware_adapter.py b/python_repositories/adapters/connection_aware_adapter.py new file mode 100644 index 0000000..6b960d4 --- /dev/null +++ b/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}") diff --git a/python_repositories/adapters/minio_adapter.py b/python_repositories/adapters/minio_adapter.py index 579c2be..bf2d34e 100644 --- a/python_repositories/adapters/minio_adapter.py +++ b/python_repositories/adapters/minio_adapter.py @@ -2,17 +2,14 @@ from __future__ import annotations import os -import structlog -import time from io import BytesIO -from typing import Self + from python_utils import check_env -from python_repositories.interfaces import ( - ConnectionAwareInterface, - ContextAwareInterface, - ObjectRepositoryInterface, +from python_repositories.adapters.connection_aware_adapter import ( + ConnectionAwareAdapter, ) +from python_repositories.interfaces import ObjectRepositoryInterface try: import minio @@ -23,11 +20,7 @@ except ImportError as exc: ) from exc -class MinioAdapter( - ObjectRepositoryInterface, - ContextAwareInterface, - ConnectionAwareInterface, -): +class MinioAdapter(ObjectRepositoryInterface, ConnectionAwareAdapter): """Minio adapter exposing basic CRUD functionality.""" endpoint_env_var_name: str = "MINIO_ENDPOINT" @@ -35,14 +28,10 @@ class MinioAdapter( secret_key_env_var_name: str = "MINIO_SECRET_KEY" bucket_env_var_name: str = "MINIO_BUCKET" chunk_size: int = 5 * 2**20 # 5 MiB - health_check_ttl_seconds: float = 1.0 + connection_name: str = "Minio" def __init__(self) -> None: - # Setup logger - self.logger = structlog.get_logger( - self.__class__.__name__, - ) - # Check environment variables + super().__init__() check_env( { self.endpoint_env_var_name, @@ -51,31 +40,11 @@ class MinioAdapter( self.bucket_env_var_name, }, ) - # Prepare internal variables self._client: minio.Minio | None = None self._bucket_name: str | None = None - 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 _is_client_ready(self) -> bool: + return self._client is not None and self._bucket_name is not None def connect(self) -> None: """Connect to the Minio server.""" @@ -119,10 +88,6 @@ class MinioAdapter( self._bucket_name = None self._invalidate_health_cache() - def _invalidate_health_cache(self) -> None: - self._health_check_at = None - self._health_check_ok = False - def _probe_connection(self) -> bool: assert self._client is not None and self._bucket_name is not None try: @@ -130,24 +95,6 @@ class MinioAdapter( except Exception: # pylint: disable=broad-except return False - def is_connected(self) -> bool: - """Check if connected to the Minio server.""" - if self._client is None or self._bucket_name is None: - 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 put( self, object_name: str, @@ -163,8 +110,7 @@ class MinioAdapter( if not isinstance(content_type, str) or len(content_type) == 0: raise ValueError("content_type must be a non-empty string") # Check connection - if not self.is_connected(): - raise ConnectionError("Not connected to Minio") + self._require_connected() assert self._client is not None and self._bucket_name is not None # Prepare buffer for reading num_bytes = data.getbuffer().nbytes @@ -189,8 +135,7 @@ class MinioAdapter( if not isinstance(object_name, str) or len(object_name) == 0: raise ValueError("object_name must be a non-empty string") # Check connection - if not self.is_connected(): - raise ConnectionError("Not connected to Minio") + self._require_connected() assert self._client is not None and self._bucket_name is not None # Get data from bucket # N.B. bucket name is set when connecting @@ -225,8 +170,7 @@ class MinioAdapter( if not isinstance(object_name, str) or len(object_name) == 0: raise ValueError("object_name must be a non-empty string") # Check connection - if not self.is_connected(): - raise ConnectionError("Not connected to Minio") + self._require_connected() assert self._client is not None and self._bucket_name is not None # Delete object from bucket # N.B. bucket name is set when connecting @@ -245,8 +189,7 @@ class MinioAdapter( raise ValueError("prefix must be a string") # Check connection # N.B. bucket name is set when connecting - if not self.is_connected(): - raise ConnectionError("Not connected to Minio") + self._require_connected() assert self._client is not None and self._bucket_name is not None # List objects in bucket objects = self._client.list_objects( diff --git a/python_repositories/adapters/redis_adapter.py b/python_repositories/adapters/redis_adapter.py index 4af6db7..6920d97 100644 --- a/python_repositories/adapters/redis_adapter.py +++ b/python_repositories/adapters/redis_adapter.py @@ -1,18 +1,15 @@ """Definition of RedisAdapter class.""" from __future__ import annotations -from typing import Self, cast +from typing import cast import os -import structlog -import time from python_utils import check_env -from python_repositories.interfaces import ( - ConnectionAwareInterface, - ContextAwareInterface, - JsonRepositoryInterface, +from python_repositories.adapters.connection_aware_adapter import ( + ConnectionAwareAdapter, ) +from python_repositories.interfaces import JsonRepositoryInterface try: import redis @@ -24,50 +21,22 @@ except ImportError as exc: ) from exc -class RedisAdapter( - JsonRepositoryInterface, - ContextAwareInterface, - ConnectionAwareInterface, -): +class RedisAdapter(JsonRepositoryInterface, ConnectionAwareAdapter): """Redis adapter exposing basic CRUD functionality.""" uri_env_var_name: str = "REDIS_URI" path: str = "." # JSON root path, updated in __init__ encoding: str = "UTF-8" - health_check_ttl_seconds: float = 1.0 + connection_name: str = "Redis" def __init__(self) -> None: - # Setup logger - self.logger = structlog.get_logger( - self.__class__.__name__, - ) - # Check environment variables + super().__init__() check_env(self.uri_env_var_name) - # Prepare internal variables self._client: redis.Redis | None = None self.path: str = RedisPath.root_path() - 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 _is_client_ready(self) -> bool: + return self._client is not None def connect(self) -> None: """Connect to the Redis server.""" @@ -100,10 +69,6 @@ class RedisAdapter( self._client = None self._invalidate_health_cache() - def _invalidate_health_cache(self) -> None: - self._health_check_at = None - self._health_check_ok = False - def _probe_connection(self) -> bool: assert self._client is not None try: @@ -111,24 +76,6 @@ class RedisAdapter( except (redis.ConnectionError, redis.TimeoutError): return False - def is_connected(self) -> bool: - """Check if connected to the Redis server.""" - if self._client is None: - 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 set(self, key: str, data: dict) -> None: """Set a JSON object in Redis.""" # Check input @@ -137,8 +84,7 @@ class RedisAdapter( if not isinstance(data, dict) or len(data) == 0: raise ValueError("Data must be a non-empty dictionary") # Check connection - if not self.is_connected(): - raise ConnectionError("Not connected to Redis") + self._require_connected() assert self._client is not None # Set data self._client.json().set(key, self.path, data) @@ -150,8 +96,7 @@ class RedisAdapter( if not isinstance(key, str) or len(key) == 0: raise ValueError("Key must be a non-empty string") # Check connection - if not self.is_connected(): - raise ConnectionError("Not connected to Redis") + self._require_connected() assert self._client is not None # Get data data = cast( @@ -167,8 +112,7 @@ class RedisAdapter( if not isinstance(key, str) or len(key) == 0: raise ValueError("Key must be a non-empty string") # Check connection - if not self.is_connected(): - raise ConnectionError("Not connected to Redis") + self._require_connected() assert self._client is not None # Delete data self._client.json().delete(key) @@ -180,8 +124,7 @@ class RedisAdapter( if not isinstance(pattern, str) or len(pattern) == 0: raise ValueError("Pattern must be a non-empty string") # Check connection - if not self.is_connected(): - raise ConnectionError("Not connected to Redis") + self._require_connected() assert self._client is not None # List keys keys_raw = cast( diff --git a/tests/unit/connection_health_test.py b/tests/unit/connection_health_test.py index 1dbccbb..8b09a06 100644 --- a/tests/unit/connection_health_test.py +++ b/tests/unit/connection_health_test.py @@ -53,7 +53,7 @@ class TestRedisConnectionHealth: redis_adapter._client = mock_client with patch( - "python_repositories.adapters.redis_adapter.time.monotonic", + "python_repositories.adapters.connection_aware_adapter.time.monotonic", return_value=100.0, ): assert redis_adapter.is_connected() @@ -67,7 +67,7 @@ class TestRedisConnectionHealth: redis_adapter._client = mock_client with patch( - "python_repositories.adapters.redis_adapter.time.monotonic", + "python_repositories.adapters.connection_aware_adapter.time.monotonic", side_effect=[100.0, 102.0], ): assert redis_adapter.is_connected() @@ -81,7 +81,7 @@ class TestRedisConnectionHealth: redis_adapter._client = mock_client with patch( - "python_repositories.adapters.redis_adapter.time.monotonic", + "python_repositories.adapters.connection_aware_adapter.time.monotonic", return_value=100.0, ): assert redis_adapter.is_connected() @@ -90,7 +90,7 @@ class TestRedisConnectionHealth: redis_adapter._client = mock_client with patch( - "python_repositories.adapters.redis_adapter.time.monotonic", + "python_repositories.adapters.connection_aware_adapter.time.monotonic", return_value=100.0, ): assert redis_adapter.is_connected() @@ -136,7 +136,7 @@ class TestMinioConnectionHealth: minio_adapter._bucket_name = "test-bucket" with patch( - "python_repositories.adapters.minio_adapter.time.monotonic", + "python_repositories.adapters.connection_aware_adapter.time.monotonic", return_value=100.0, ): assert minio_adapter.is_connected() @@ -151,7 +151,7 @@ class TestMinioConnectionHealth: minio_adapter._bucket_name = "test-bucket" with patch( - "python_repositories.adapters.minio_adapter.time.monotonic", + "python_repositories.adapters.connection_aware_adapter.time.monotonic", side_effect=[100.0, 102.0], ): assert minio_adapter.is_connected() @@ -166,7 +166,7 @@ class TestMinioConnectionHealth: minio_adapter._bucket_name = "test-bucket" with patch( - "python_repositories.adapters.minio_adapter.time.monotonic", + "python_repositories.adapters.connection_aware_adapter.time.monotonic", return_value=100.0, ): assert minio_adapter.is_connected() @@ -176,7 +176,7 @@ class TestMinioConnectionHealth: minio_adapter._bucket_name = "test-bucket" with patch( - "python_repositories.adapters.minio_adapter.time.monotonic", + "python_repositories.adapters.connection_aware_adapter.time.monotonic", return_value=100.0, ): assert minio_adapter.is_connected()