Brian Bjarke JensenandCursor
67bb88fbb4
PR Title Check / check-title (pull_request) Successful in 5s
Test Python Package / unit-tests (pull_request) Successful in 13s
Test Python Package / integration-tests (pull_request) Successful in 30s
Test Python Package / coverage-report (pull_request) Failing after 15s
Code Quality Pipeline / code-quality (pull_request) Successful in 53s
Move shared connect orchestration into the base adapter so both backends short-circuit when already healthy and only reconnect after a failed probe. Co-authored-by: Cursor <[email protected]>
103 lines
3.4 KiB
Python
103 lines
3.4 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._client_injected = False
|
|
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."""
|
|
|
|
@abstractmethod
|
|
def _validate_injected_client(self) -> None:
|
|
"""Verify an injected client is reachable; raise ConnectionError on failure."""
|
|
|
|
@abstractmethod
|
|
def _establish_connection(self) -> None:
|
|
"""Create a backend client and set internal connection state."""
|
|
|
|
def connect(self) -> None:
|
|
"""Connect to the backend; idempotent when already connected and healthy."""
|
|
if self._client_injected:
|
|
self._validate_injected_client()
|
|
self._invalidate_health_cache()
|
|
return
|
|
if self._is_client_ready() and self.is_connected():
|
|
self.logger.info(f"Already connected to {self.connection_name}")
|
|
return
|
|
self.disconnect()
|
|
self._establish_connection()
|
|
self._invalidate_health_cache()
|
|
|
|
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}")
|