Source code for tidalpy.errors

"""Structured exception hierarchy for tidalpy.

Every error carries a machine-readable :meth:`to_dict` payload so hosts
(Django models, Celery task rows, logs) can persist failures without
parsing human-readable text. The legacy integration had to guess what
happened from CLI exit codes and stderr fragments; this hierarchy makes
the failure mode explicit and testable.

Error taxonomy
--------------
- :class:`ConfigError`      bad configuration, fail fast before network I/O
- :class:`AuthError`        missing/expired/unusable credentials
- :class:`RateLimitError`   TIDAL kept answering 429 beyond the wait budget
- :class:`NotFoundError`    resource id does not exist (either API version)
- :class:`RegionLockedError` content unavailable in the account's country
- :class:`PreviewOnlyError` account can only hear 30 s previews (free tier)
- :class:`QualityUnavailableError` no stream at any attempted tier
- :class:`StreamError`      playback resolution failed for other reasons
- :class:`ManifestError`    manifest missing, corrupt, or DRM-protected
- :class:`UnsupportedType`  resource kind this library does not handle
- :class:`DownloadError`    network/disk failure while fetching media bytes
- :class:`DiskFullError`    the disk could not accept the media bytes
- :class:`CancelledError`   the host cancelled the operation
"""

from __future__ import annotations

from datetime import datetime, timezone
from typing import Any


def _utc_now_iso() -> str:
    return datetime.now(timezone.utc).isoformat()




class TidalError(Exception):
    """Base class for every error raised by tidalpy.

    Every error carries a stable machine ``code`` (e.g.
    ``"auth-expired"``) that hosts may key off without string-matching
    the message, and a ``retryable`` flag: ``True`` when the same call
    may succeed later without host action (transient network, rate
    limit), ``False`` for permanent failures (bad token, region lock)
    where retrying wastes time.
    """

    code: str = "tidal-error"
    retryable: bool = False
    # Optional context (subclasses assign these unconditionally; the base
    # class only sets them when provided, so ``to_dict`` uses getattr).
    status_code: int | None
    sub_status: int | str | None



    def __init__(
        self,
        message: str,
        *details: Any,
        status_code: int | None = None,
        sub_status: int | str | None = None,
    ) -> None:
        super().__init__(message)
        self.message = message
        self.details: list[Any] = list(details)
        if status_code is not None:
            self.status_code = status_code
        if sub_status is not None:
            self.sub_status = sub_status




    def to_dict(self) -> dict[str, Any]:
        """JSON-serialisable representation (safe for Django JSONField)."""
        payload: dict[str, Any] = {
            "error": type(self).__name__,
            "code": self.code,
            "message": str(self),
            "retryable": self.retryable,
            "timestamp": _utc_now_iso(),
        }
        status_code = getattr(self, "status_code", None)
        sub_status = getattr(self, "sub_status", None)
        if status_code is not None:
            payload["status_code"] = status_code
        if sub_status is not None:
            payload["sub_status"] = sub_status
        if self.details:
            payload["details"] = [str(d) for d in self.details]
        return payload






class ConfigError(TidalError):
    """Configuration is missing or inconsistent."""

    code = "config"





class AuthError(TidalError):
    """Credentials are missing, expired, or unusable for this client."""

    code = "auth"





class RateLimitError(TidalError):
    """TIDAL returned 429 repeatedly; the wait budget is exhausted."""

    code = "rate-limited"
    retryable = True





class NotFoundError(TidalError):
    """The resource id does not exist (v1 and, when applicable, v2)."""

    code = "not-found"





class RegionLockedError(TidalError):
    """The content is not available in the account's country."""

    code = "region-locked"





class PreviewOnlyError(TidalError):
    """Only a preview (e.g. free account) is offered for this content."""

    code = "preview-only"





class QualityUnavailableError(TidalError):
    """No stream could be resolved at any attempted quality tier."""

    code = "quality-unavailable"





class TidalApiError(TidalError):
    """A Tidal API rejection with HTTP/subStatus context attached."""

    code = "api"



    def __init__(
        self,
        message: str,
        status_code: int | None = None,
        sub_status: int | str | None = None,
        codes: list[str] | None = None,
        *details: Any,
    ) -> None:
        super().__init__(
            message, *details, status_code=status_code, sub_status=sub_status
        )
        self.status_code = status_code
        self.sub_status = sub_status
        # Machine codes the error body carried (e.g.
        # ``CLIENT_NOT_ENTITLED``): account-capability rejections are
        # distinguishable from track-specific blocks.
        self.codes: list[str] = list(codes or [])






class StreamError(TidalError):
    """Playback info / stream resolution failed for other reasons."""

    code = "stream"





class ManifestError(StreamError):
    """A stream manifest is missing, corrupt, oversized, or DRM-protected."""

    code = "manifest"





class UnsupportedType(TidalError):
    """The resource kind is not handled (e.g. videos) or not requested."""

    code = "unsupported-type"





class DownloadError(TidalError):
    """Fetching the media bytes failed (network, truncation, disk)."""

    code = "download"
    retryable = True





class DiskFullError(DownloadError):
    """The destination disk ran out of space mid-download."""

    code = "disk-full"
    retryable = False





class CancelledError(TidalError):
    """The host cancelled the in-flight operation."""

    code = "cancelled"