Brian Bjarke JensenandCursor
7b950e0d93
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]>
81 lines
2.5 KiB
Python
81 lines
2.5 KiB
Python
"""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}")
|