lille-vemmelund/python-repositories
2
0
Fork 0
Code Issues 3 Pull Requests 1 Actions Packages Releases 17 Activity

Extract shared connection lifecycle into ConnectionAwareAdapter.

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