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 <cursoragent@cursor.com>
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
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(
python_repositories/adapters/redis_adapter.py
+13 -70
View File
@@ -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(
tests/unit/connection_health_test.py
+8 -8
View File
@@ -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()