product-54354138-1/tools/errors.py

"""Typed errors raised deliberately by the toolkit.

The image-source layer never degrades a failed or unknown source into a silent,
empty result.  Every such situation raises a :class:`ToolkitError` subclass that
carries a stable ``code`` string plus machine-readable ``details`` so the CLI can
map it onto an exit status and a structured JSON envelope.

Hierarchy::

    ToolkitError
    +-- SourceError
        +-- UnknownSourceError        (unknown source id, lists available ids)
        +-- DuplicateSourceError      (registry already holds that id)
        +-- SearchRequestError        (topic/limit rejected before any request)
        +-- SourceAccessError         (the source could not be used)
            +-- SourceUnreachableError    (transport/DNS/timeout failure)
            +-- SourceHTTPError           (non-retryable or exhausted HTTP status)
            +-- SourcePayloadError        (body is not the expected JSON shape)
            +-- HostNotAllowedError       (URL outside the approved hosts)
"""

from __future__ import annotations

from typing import Any, Iterable, Mapping

__all__ = [
    "DuplicateSourceError",
    "HostNotAllowedError",
    "SearchRequestError",
    "SourceAccessError",
    "SourceError",
    "SourceHTTPError",
    "SourcePayloadError",
    "SourceUnreachableError",
    "ToolkitError",
    "UnknownSourceError",
]


class ToolkitError(Exception):
    """Base class for errors the toolkit raises on purpose."""

    code = "toolkit_error"

    def __init__(
        self,
        message: str,
        *,
        code: str | None = None,
        details: Mapping[str, Any] | None = None,
    ) -> None:
        super().__init__(message)
        self.message = message
        if code is not None:
            self.code = code
        self.details: dict[str, Any] = dict(details or {})

    def to_dict(self) -> dict[str, Any]:
        """Return the JSON-safe envelope fragment describing this error."""

        payload: dict[str, Any] = {"code": self.code, "message": self.message}
        if self.details:
            payload["details"] = dict(self.details)
        return payload

    def __str__(self) -> str:  # pragma: no cover - trivial
        return self.message


class SourceError(ToolkitError):
    """Base class for every image-source problem."""

    code = "source_error"


class UnknownSourceError(SourceError):
    """Raised when the caller asks for a source id the registry does not know."""

    code = "source_unknown"

    def __init__(self, source_id: str, available: Iterable[str]) -> None:
        self.source_id = str(source_id)
        self.available: tuple[str, ...] = tuple(sorted(str(item) for item in available))
        listing = ", ".join(self.available) if self.available else "<none>"
        super().__init__(
            f"unknown image source {self.source_id!r}; available sources: {listing}",
            details={"source_id": self.source_id, "available_sources": list(self.available)},
        )


class DuplicateSourceError(SourceError):
    """Raised when an adapter id is registered twice."""

    code = "source_duplicate"

    def __init__(self, source_id: str) -> None:
        self.source_id = str(source_id)
        super().__init__(
            f"image source {self.source_id!r} is already registered",
            details={"source_id": self.source_id},
        )


class SearchRequestError(SourceError):
    """Raised when the caller supplies a topic/limit a source cannot honour."""

    code = "search_invalid_request"

    def __init__(self, message: str, *, details: Mapping[str, Any] | None = None) -> None:
        super().__init__(message, details=details)


class SourceAccessError(SourceError):
    """Raised when a known source cannot be reached or refuses to answer."""

    code = "source_access"

    def __init__(
        self,
        message: str,
        *,
        url: str | None = None,
        source_id: str | None = None,
        details: Mapping[str, Any] | None = None,
    ) -> None:
        merged: dict[str, Any] = dict(details or {})
        if url is not None:
            merged["url"] = url
        if source_id is not None:
            merged["source_id"] = source_id
        super().__init__(message, details=merged)
        self.url = url
        self.source_id = source_id


class SourceUnreachableError(SourceAccessError):
    """Raised when the transport cannot deliver a request to the host at all."""

    code = "source_unreachable"

    def __init__(self, url: str, *, source_id: str | None = None, reason: str | None = None) -> None:
        message = f"image source host is unreachable: {url}"
        if reason:
            message = f"{message} ({reason})"
        super().__init__(message, url=url, source_id=source_id, details={"reason": reason} if reason else None)


class SourceHTTPError(SourceAccessError):
    """Raised for an HTTP status the client cannot recover from."""

    code = "source_http_error"

    def __init__(self, url: str, status: int, *, source_id: str | None = None) -> None:
        super().__init__(
            f"image source returned HTTP status {status}: {url}",
            url=url,
            source_id=source_id,
            details={"status": status},
        )
        self.status = int(status)


class SourcePayloadError(SourceAccessError):
    """Raised when a source answers with something the adapter cannot read."""

    code = "source_invalid_response"

    def __init__(self, url: str, *, source_id: str | None = None, reason: str | None = None) -> None:
        message = f"image source returned an unusable response: {url}"
        if reason:
            message = f"{message} ({reason})"
        super().__init__(message, url=url, source_id=source_id, details={"reason": reason} if reason else None)


class HostNotAllowedError(SourceAccessError):
    """Raised when a URL leaves the approved HTTPS host allowlist."""

    code = "source_host_not_allowed"

    def __init__(self, url: str, *, allowed: Iterable[str] = ()) -> None:
        self.allowed: tuple[str, ...] = tuple(sorted(str(item) for item in allowed))
        super().__init__(
            f"refusing to contact {url!r}: only approved HTTPS hosts are allowed "
            f"({', '.join(self.allowed) if self.allowed else 'none configured'})",
            url=str(url),
            details={"allowed_hosts": list(self.allowed)},
        )


# ---------------------------------------------------------------------------
# Request errors: invalid command lines and JSON request documents
# ---------------------------------------------------------------------------

#: Image formats the CLI accepts.
ACCEPTED_FORMATS: tuple[str, ...] = ("png", "ico")


class RequestError(ToolkitError):
    """Base class for every unusable request (CLI flags or request document)."""

    code = "request_invalid"

    def __init__(
        self,
        message: str,
        *,
        code: str | None = None,
        details: Mapping[str, Any] | None = None,
        field: str | None = None,
        accepted_fields: Iterable[str] | None = None,
        accepted_formats: Iterable[str] | None = None,
    ) -> None:
        merged: dict[str, Any] = dict(details or {})
        if field is not None:
            merged["field"] = field
        if accepted_fields is not None:
            merged["accepted_fields"] = list(accepted_fields)
        if accepted_formats is not None:
            merged["accepted_formats"] = list(accepted_formats)
        super().__init__(message, code=code, details=merged)
        self.field = field
        self.accepted_fields: tuple[str, ...] = tuple(accepted_fields or ())
        self.accepted_formats: tuple[str, ...] = tuple(accepted_formats or ())


class RequestDocumentError(RequestError):
    """Raised when the JSON request document cannot be read or decoded."""

    code = "request_document"


class RequestFieldError(RequestError):
    """Raised when a single request field is missing, unknown or malformed."""

    code = "request_field"


class UnsupportedFormatError(RequestError):
    """Raised for a format outside :data:`ACCEPTED_FORMATS`."""

    code = "request_format_unsupported"

    def __init__(self, value: Any, accepted: Iterable[str] = ACCEPTED_FORMATS) -> None:
        accepted = tuple(accepted)
        self.value = str(value)
        super().__init__(
            f"unsupported image format {self.value!r}; "
            f"accepted formats: {', '.join(accepted)}",
            field="format",
            accepted_formats=accepted,
            details={"requested_format": self.value},
        )


__all__ += [
    "ACCEPTED_FORMATS",
    "RequestDocumentError",
    "RequestError",
    "RequestFieldError",
    "UnsupportedFormatError",
]