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:
        """Return whether all healthchecks passed (or allowed partial failure)."""
        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,  # noqa: FBT001
        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)

@dataclass(frozen=True)
class ProbeRunner:
    """Immutable runner that executes probes according to RunPolicy.

    Use as an async context manager to ensure proper resource cleanup:

        async with ProbeRunner(policy) as runner:
            report = await runner.run(probe)

    Attributes:
        policy: RunPolicy instance controlling execution behavior.
        _probes: Internal list of probes that have been registered with this runner.
        _probe_ids: Internal set of probe object IDs for deduplication.
    """

    policy: RunPolicy = field(default_factory=RunPolicy)
    _probes: list[Probe] = field(default_factory=list, init=False, repr=False, compare=False)
    _probe_ids: set[int] = field(default_factory=set, init=False, repr=False, compare=False)

    async def __aenter__(self) -> ProbeRunner:  # noqa: PYI034
        """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 probes on context exit."""
        await self.close()

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

        Args:
            probe: Probe with checks to execute.

        Returns:
            HealthCheckReport for the probe run.
        """
        probe_id = id(probe)
        if probe_id not in self._probe_ids:
            self._probe_ids.add(probe_id)
            self._probes.append(probe)

        timeout = None
        if self.policy.probe_timeout_ms is not None:
            timeout = self.policy.probe_timeout_ms / 1000

        on_check_start = _noop_on_check_start if self.policy.execution == "sequential" else None

        report = await run_probe(
            probe,
            timeout=timeout,
            on_check_start=on_check_start,
            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 resources for probes executed by this runner."""
        await close_probes(self._probes)

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

    Attributes:
        mode: Controls how probe failures affect overall health status.
            - "strict": Any failure marks the health check as failed
            - "reporting": Failures are reported but don't fail the check
        execution: Controls how probes are executed.
            - "parallel": All probes run concurrently
            - "sequential": Probes run one at a time
        probe_timeout_ms: Optional timeout in milliseconds for each probe.
            If None, probes use their default timeout.
        health_evaluation: Controls evaluation strategy.
            - "all_required": All probes must pass for overall health
            - "partial_allowed": Some probes can fail without failing overall
    """

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

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

        Raises:
            ValueError: If mode/execution/health_evaluation is invalid or timeout is non-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)

@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."""

    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"
    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."""

    hosts: list[str] = field(default_factory=lambda: ["localhost:9200"])
    http_auth: tuple[str, str] | None = None
    use_ssl: bool = False
    verify_certs: bool = False
    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 | 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 intended for local development only.
    Do not use these defaults in production or on non-local brokers; set
    explicit credentials or use a secrets manager. See SECURITY.md.
    """

    host: str = "localhost"
    port: int = 5672
    user: str = "guest"
    password: str = "guest"  # noqa: S105
    vhost: str = "/"
    secure: bool = False
    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 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)

@dataclass(frozen=True)
class UrlConfig:
    """Configuration for URL health check.

    Use only trusted URLs from application configuration; do not pass
    user-controlled input to avoid SSRF. Validation and behaviour are
    provided by :func:`~fast_healthchecks.utils.validate_url_ssrf` and
    :func:`~fast_healthchecks.utils.validate_host_ssrf_async`. See the
    SSRF documentation in the docs.
    """

    url: str = ""
    username: str | None = None
    password: str | None = None
    verify_ssl: bool = True
    follow_redirects: bool = True
    timeout: float = DEFAULT_HC_TIMEOUT
    block_private_hosts: bool = False

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

class HealthCheck(Protocol[T_co]):
    """Base class for health checks."""

    async def __call__(self) -> T_co: ...

class HealthCheckDSN(ConfigDictMixin, HealthCheck[T_co], Generic[T_co, T_parsed]):
    """Base class for health checks that can be created from a DSN.

    Contract: subclasses must define _allowed_schemes(), _default_name(),
    parse_dsn(), and _from_parsed_dsn(). The check stores its display name in
    _name (used in HealthCheckResult and error reporting). DSN validation uses
    validate_dsn(); fast_healthchecks.dsn NewTypes are typing-only, not runtime.

    Type parameters: T_co is the result type (e.g. HealthCheckResult);
    T_parsed is the type returned by parse_dsn() and accepted by _from_parsed_dsn().
    """

    @classmethod
    @abstractmethod
    def _allowed_schemes(cls) -> tuple[str, ...]:
        """Return DSN schemes allowed for this check (e.g. ('redis', 'rediss'))."""
        raise NotImplementedError  # pragma: no cover

    @classmethod
    @abstractmethod
    def _default_name(cls) -> str:
        """Return the default check name for from_dsn (e.g. 'Redis')."""
        raise NotImplementedError  # pragma: no cover

    @classmethod
    @abstractmethod
    def parse_dsn(cls, dsn: str) -> T_parsed:
        """Parse the DSN string. Subclasses must implement."""
        raise NotImplementedError  # pragma: no cover

    @classmethod
    @abstractmethod
    def _from_parsed_dsn(
        cls,
        parsed: T_parsed,
        *,
        name: str = "Service",
        timeout: float = DEFAULT_HC_TIMEOUT,
        **kwargs: object,
    ) -> HealthCheckDSN[T_co, T_parsed]:
        """Create a check instance from parsed DSN."""
        raise NotImplementedError  # pragma: no cover

    @classmethod
    def from_dsn(
        cls,
        dsn: str,
        *,
        name: str | None = None,
        timeout: float = DEFAULT_HC_TIMEOUT,
        **kwargs: object,
    ) -> HealthCheckDSN[T_co, T_parsed]:
        """Create a check instance from a DSN string.

        Returns:
            HealthCheckDSN: Configured check instance.
        """
        if name is None:
            name = cls._default_name()
        dsn = cls.validate_dsn(dsn, allowed_schemes=cls._allowed_schemes())
        parsed = cls.parse_dsn(dsn)
        return cls._from_parsed_dsn(parsed, name=name, timeout=timeout, **kwargs)

    @classmethod
    def validate_dsn(cls, dsn: str, *, allowed_schemes: tuple[str, ...]) -> str:
        """Validate the DSN has an allowed scheme.

        Allows compound schemes (e.g. postgresql+asyncpg) when the base
        part before '+' is in allowed_schemes. Scheme comparison is case-insensitive.

        Returns:
            str: The DSN string (stripped of leading/trailing whitespace).

        Raises:
            TypeError: If dsn is not a string.
            ValueError: If DSN is empty or scheme is not in allowed_schemes.
        """
        if not isinstance(dsn, str):
            msg = f"DSN must be str, got {type(dsn).__name__!r}"
            raise TypeError(msg) from None

        dsn = dsn.strip()
        if not dsn:
            msg = "DSN cannot be empty"
            raise ValueError(msg) from None

        if not allowed_schemes:
            msg = "allowed_schemes cannot be empty"
            raise ValueError(msg) from None

        parsed = urlsplit(dsn)
        scheme = (parsed.scheme or "").lower()
        base_scheme = scheme.split("+", 1)[0] if "+" in scheme else scheme

        allowed_set = frozenset(s.lower() for s in allowed_schemes)
        if scheme not in allowed_set and base_scheme not in allowed_set:
            schemes_str = ", ".join(sorted(allowed_set))
            msg = f"DSN scheme must be one of {schemes_str} (or compound e.g. postgresql+driver), got {scheme!r}"
            raise ValueError(msg) from None

        return dsn