API Reference¶

class HealthCheckError(Exception):
    """Base exception for health-check-related failures.

    Raised or used as a base for timeouts, SSRF validation, and other
    health-check errors. Subclasses preserve the original exception type
    (e.g. HealthCheckTimeoutError is also an asyncio.TimeoutError) so
    existing code that catches TimeoutError or ValueError continues to work.
    """

@dataclass(frozen=True)
class HealthCheckReport:
    """Report of healthchecks.

    Attributes:
        results: List of healthcheck results.
        allow_partial_failure: If True, report is healthy when at least one check passes.
    """

    results: list[HealthCheckResult]
    allow_partial_failure: bool = False

    def __str__(self) -> str:
        """Return a string representation of the report."""
        return "\n".join(str(result) for result in self.results)

    @property
    def healthy(self) -> bool:
        """Whether all health checks passed or partial failure was allowed."""
        if self.allow_partial_failure:
            return any(result.healthy for result in self.results)
        return all(result.healthy for result in self.results)

@dataclass(frozen=True, init=False)
class HealthCheckResult:
    """Result of a healthcheck.

    Attributes:
        name: Name of the healthcheck.
        healthy: Whether the healthcheck passed.
        error: Structured error details if the healthcheck failed.
    """

    name: str
    healthy: bool
    error: HealthError | None = None

    def __init__(
        self,
        name: str,
        healthy: bool,  # ruff: ignore[boolean-type-hint-positional-argument] - public keyword flag kept for API compatibility
        error: HealthError | None = None,
        *,
        error_details: str | None = None,
    ) -> None:
        """Create a health check result.

        The ``error_details`` keyword is accepted for backward compatibility
        and is converted into ``error.message``.

        Raises:
            ValueError: If both ``error`` and ``error_details`` are provided.
        """
        if error is not None and error_details is not None:
            msg = "Provide either error or error_details, not both"
            raise ValueError(msg)

        resolved_error = error
        if resolved_error is None and error_details is not None:
            resolved_error = HealthError(code="CHECK_EXCEPTION", message=error_details)

        object.__setattr__(self, "name", name)
        object.__setattr__(self, "healthy", healthy)
        object.__setattr__(self, "error", resolved_error)

    @property
    def error_details(self) -> str | None:
        """Backward-compatible error message accessor."""
        if self.error is None:
            return None
        return self.error.message

    def __str__(self) -> str:
        """Return a string representation of the result."""
        return f"{self.name}: {'healthy' if self.healthy else 'unhealthy'}"

class HealthCheckSSRFError(HealthCheckError, ValueError):
    """Raised when URL or host validation fails (SSRF / block_private_hosts).

    Subclass of both HealthCheckError and ValueError so that
    ``except ValueError`` still catches it.
    """

class HealthCheckTimeoutError(HealthCheckError, asyncio.TimeoutError):
    """Raised when a probe or check run exceeds its timeout.

    Subclass of both HealthCheckError and asyncio.TimeoutError so that
    ``except asyncio.TimeoutError`` or ``except TimeoutError`` still catch it.
    """

    code: str

    def __init__(self, message: str = "Probe timed out", *, code: str = "PROBE_TIMEOUT") -> None:
        """Create timeout error with machine-readable timeout code."""
        super().__init__(message)
        self.code = code

@dataclass(frozen=True)
class HealthError:
    """Machine-readable error details for failed health checks.

    Attributes:
        code: Machine-readable error code (e.g., "PROBE_TIMEOUT", "CHECK_EXCEPTION").
        message: Human-readable error message describing what went wrong.
        duration_ms: Time in milliseconds that the probe took to execute.
        timeout_ms: Timeout limit in milliseconds that was applied to the probe.
        meta: Additional metadata about the error (e.g., exception type, URL).
            Secrets are automatically redacted from this field.
    """

    code: str
    message: str
    duration_ms: int = 0
    timeout_ms: int | None = None
    meta: dict[str, object] = field(default_factory=dict)

    def __post_init__(self) -> None:
        """Store a redacted copy so secrets are safe before serialization or logging."""
        object.__setattr__(self, "meta", redact_secrets_in_dict(self.meta))

class Probe(NamedTuple):
    """A named sequence of health checks executed as one probe."""

    name: str
    checks: Sequence[Check]
    summary: str | None = None
    default_check_timeout_ms: int | None = None

    @property
    def endpoint_summary(self) -> str:
        """The explicit summary or a readable summary derived from the route."""
        if self.summary:
            return self.summary
        title = re.sub(
            pattern=r"[^a-z0-9]+",
            repl=" ",
            string=self.name.lower().capitalize(),
            flags=re.IGNORECASE,
        )
        return f"{title} probe"

@dataclass(frozen=True)
class ProbeRunner:
    """Execute probes and close only resource-owning checks seen by the runner.

    The runner owns the lifecycle of resource checks it has seen: ``close()``
    and resource tracking are serialized by an internal lock so a concurrent
    ``close()`` cannot race registration. Probe execution itself is not
    locked; concurrent ``run()`` calls proceed in parallel.
    """

    policy: RunPolicy = field(default_factory=RunPolicy)
    _resource_checks: list[object] = field(default_factory=list, init=False, repr=False, compare=False)
    _resource_check_ids: set[int] = field(default_factory=set, init=False, repr=False, compare=False)
    _resources_lock: asyncio.Lock = field(default_factory=asyncio.Lock, init=False, repr=False, compare=False)

    async def __aenter__(self) -> ProbeRunner:  # ruff: ignore[non-self-return-type] - returns the concrete subclass by design
        """Return self for async context-manager usage."""
        return self

    async def __aexit__(
        self,
        _exc_type: type[BaseException] | None,
        _exc: BaseException | None,
        _tb: TracebackType | None,
    ) -> None:
        """Always close managed checks on context exit."""
        await self.close()

    def _track_resources(self, probe: Probe) -> None:
        for check in probe.checks:
            check_id = id(check)
            if check_id in self._resource_check_ids or not callable(getattr(check, "aclose", None)):
                continue
            self._resource_check_ids.add(check_id)
            self._resource_checks.append(check)

    async def run(self, probe: Probe) -> HealthCheckReport:
        """Run probe checks and return a report.

        Returns:
            The evaluated health-check report.
        """
        async with self._resources_lock:
            self._track_resources(probe)
        timeout = None if self.policy.probe_timeout_ms is None else self.policy.probe_timeout_ms / 1000
        report = await run_probe(
            probe,
            timeout=timeout,
            execution=self.policy.execution,
            max_concurrency=self.policy.max_concurrency,
            on_timeout_return_failure=self.policy.mode == "reporting",
        )
        if self.policy.health_evaluation == "partial_allowed":
            return HealthCheckReport(results=report.results, allow_partial_failure=True)
        return report

    async def close(self) -> None:
        """Close resource-owning checks observed by this runner."""
        async with self._resources_lock:
            if await _close_checks(self._resource_checks):
                await asyncio.sleep(0.1)
            self._resource_checks.clear()
            self._resource_check_ids.clear()

@dataclass(frozen=True)
class RunPolicy:
    """Immutable policy controlling probe execution behavior.

    ``max_concurrency`` caps how many checks run at once in parallel mode
    (default 8); ``None`` removes the cap.
    """

    mode: RunMode = "strict"
    execution: ExecutionMode = "parallel"
    probe_timeout_ms: int | None = None
    health_evaluation: HealthEvaluationMode = "all_required"
    max_concurrency: int | None = DEFAULT_MAX_CONCURRENCY

    def __post_init__(self) -> None:
        """Validate policy values.

        Raises:
            ValueError: If a mode is unknown, the timeout is not positive, or
                max_concurrency is not positive.
        """
        if self.mode not in _VALID_RUN_MODES:
            msg = f"Invalid run mode: {self.mode}"
            raise ValueError(msg)
        if self.execution not in _VALID_EXECUTION_MODES:
            msg = f"Invalid execution mode: {self.execution}"
            raise ValueError(msg)
        if self.health_evaluation not in _VALID_HEALTH_EVALUATIONS:
            msg = f"Invalid health evaluation mode: {self.health_evaluation}"
            raise ValueError(msg)
        if self.probe_timeout_ms is not None and self.probe_timeout_ms <= 0:
            msg = "probe_timeout_ms must be > 0 when provided"
            raise ValueError(msg)
        if self.max_concurrency is not None and self.max_concurrency <= 0:
            msg = "max_concurrency must be > 0 when provided"
            raise ValueError(msg)

@dataclass(frozen=True)
class FunctionConfig:
    """Configuration for function health check."""

    args: tuple[object, ...] = ()
    kwargs: dict[str, object] | None = None
    timeout: float = DEFAULT_HC_TIMEOUT

    def to_dict(self) -> dict[str, Any]:
        """Return config as a dict for serialization."""
        d = asdict(self)
        d["args"] = list(d.get("args") or ())
        d["kwargs"] = dict(d["kwargs"]) if d.get("kwargs") else {}
        return d

@dataclass(frozen=True)
class KafkaConfig:
    """Configuration for Kafka health check."""

    bootstrap_servers: str = "localhost:9092"
    ssl_context: _ssl.SSLContext | None = None
    security_protocol: SecurityProtocol = "PLAINTEXT"
    sasl_mechanism: SaslMechanism = "PLAIN"
    sasl_plain_username: str | None = None
    sasl_plain_password: str | None = None
    timeout: float = DEFAULT_HC_TIMEOUT

    def __post_init__(self) -> None:
        """Validate security_protocol and sasl_mechanism.

        Raises:
            ValueError: If security_protocol or sasl_mechanism is invalid.
        """
        if self.security_protocol not in VALID_SECURITY_PROTOCOLS:
            msg = f"Invalid security protocol: {self.security_protocol}"
            raise ValueError(msg)
        if self.sasl_mechanism not in VALID_SASL_MECHANISMS:
            msg = f"Invalid SASL mechanism: {self.sasl_mechanism}"
            raise ValueError(msg)

    def to_dict(self) -> dict[str, Any]:
        """Return config as a dict for serialization."""
        # ssl_context is not deepcopy-safe; build dict manually
        return {
            "bootstrap_servers": self.bootstrap_servers,
            "ssl_context": self.ssl_context,
            "security_protocol": self.security_protocol,
            "sasl_mechanism": self.sasl_mechanism,
            "sasl_plain_username": self.sasl_plain_username,
            "sasl_plain_password": self.sasl_plain_password,
            "timeout": self.timeout,
        }

@dataclass(frozen=True)
class MongoConfig:
    """Configuration for MongoDB health check.

    ``tls=None`` leaves the driver default untouched; ``tls_ca_file`` is
    forwarded as ``tlsCAFile`` when set.
    """

    hosts: str | list[str] = "localhost"
    port: int | None = 27017
    user: str | None = None
    password: str | None = None
    database: str | None = None
    auth_source: str = "admin"
    tls: bool | None = None
    tls_ca_file: str | None = None
    timeout: float = DEFAULT_HC_TIMEOUT

    def to_dict(self) -> dict[str, Any]:
        """Return config as a dict for serialization."""
        return asdict(self)

@dataclass(frozen=True)
class OpenSearchConfig:
    """Configuration for OpenSearch health check.

    ``verify_certs`` defaults to ``True``: TLS connections verify the server
    certificate unless explicitly disabled. It only applies when
    ``use_ssl=True``.
    """

    hosts: list[str] = field(default_factory=lambda: ["localhost:9200"])
    http_auth: tuple[str, str] | None = None
    use_ssl: bool = False
    verify_certs: bool = True
    ssl_show_warn: bool = False
    ca_certs: str | None = None
    timeout: float = DEFAULT_HC_TIMEOUT

    def to_dict(self) -> dict[str, Any]:
        """Return config as a dict for serialization."""
        return asdict(self)

@dataclass(frozen=True)
class PostgresAsyncPGConfig:
    """Configuration for PostgreSQL health check (asyncpg driver)."""

    host: str = "localhost"
    port: int = 5432
    user: str | None = None
    password: str | None = None
    database: str | None = None
    ssl: _ssl.SSLContext | SslMode | bool | None = None
    direct_tls: bool = False
    timeout: float = DEFAULT_HC_TIMEOUT

    def to_dict(self) -> dict[str, Any]:
        """Return config as a dict for serialization."""
        # ssl is not serializable; include as None in dict for consistent keys
        return {
            "host": self.host,
            "port": self.port,
            "user": self.user,
            "password": self.password,
            "database": self.database,
            "ssl": self.ssl,
            "direct_tls": self.direct_tls,
            "timeout": self.timeout,
        }

@dataclass(frozen=True)
class PostgresPsycopgConfig:
    """Configuration for PostgreSQL health check (psycopg driver)."""

    host: str = "localhost"
    port: int = 5432
    user: str | None = None
    password: str | None = None
    database: str | None = None
    sslmode: str | None = None
    sslcert: str | None = None
    sslkey: str | None = None
    sslrootcert: str | None = None
    timeout: float = DEFAULT_HC_TIMEOUT

    def to_dict(self) -> dict[str, Any]:
        """Return config as a dict for serialization."""
        return asdict(self)

@dataclass(frozen=True)
class RabbitMQConfig:
    """Configuration for RabbitMQ health check.

    **Security:** The default ``user`` and ``password`` (``"guest"``) match
    RabbitMQ's default credentials and are accepted only for loopback hosts
    (``localhost``, ``127.0.0.1``, ``::1``). Configuring them for any other
    host raises ``ValueError``; set explicit credentials or use a secrets
    manager. See SECURITY.md.
    """

    host: str = "localhost"
    port: int = 5672
    user: str = "guest"
    password: str = "guest"  # ruff: ignore[hardcoded-password-string] - kept deliberately
    vhost: str = "/"
    secure: bool = False
    timeout: float = DEFAULT_HC_TIMEOUT

    def __post_init__(self) -> None:
        """Reject RabbitMQ default credentials for non-loopback brokers.

        Raises:
            ValueError: If both ``user`` and ``password`` are ``"guest"`` and
                ``host`` is not a loopback address.
        """
        if self.user == "guest" and self.password == "guest" and self.host not in _LOOPBACK_HOSTS:  # ruff: ignore[hardcoded-password-string] - compares against the known RabbitMQ default
            msg = (
                f"RabbitMQ default credentials 'guest'/'guest' are only valid for loopback hosts, "
                f"got host {self.host!r}; set explicit credentials"
            )
            raise ValueError(msg)

    def to_dict(self) -> dict[str, Any]:
        """Return config as a dict for serialization."""
        return asdict(self)

@dataclass(frozen=True)
class RedisConfig:
    """Configuration for Redis health check."""

    host: str = "localhost"
    port: int = 6379
    database: int | str = 0
    user: str | None = None
    password: str | None = None
    ssl: bool = False
    ssl_ca_certs: str | None = None
    timeout: float = DEFAULT_HC_TIMEOUT

    def to_dict(self) -> dict[str, Any]:
        """Return config as a dict for serialization."""
        return asdict(self)