Source code for embedl_hub._internal.tracking.errors

from __future__ import annotations

from pathlib import Path

from embedl_hub._internal.tracking.rest_api import ApiError, ErrorCode


class DomainError(Exception):
    """Base class for domain-specific exceptions."""




class TrackingFailureWarning(UserWarning):
    """Warns that a tracking operation failed and was skipped.

    Emitted instead of raising when the failing call is covered by a
    safe scope — a run started with ``safe=True``, or a logging call
    made inside :meth:`~embedl_hub.tracking.Client.safe_log`.  The
    script continues, and whether the affected data reached the Hub is
    unknown: a request that failed while the response was being read may
    already have been committed.  Treat it as uncertain rather than lost,
    since retrying on the assumption it was lost can duplicate it.

    This is its own category so it can be filtered independently of
    other warnings.  Escalating it to an error is not applied to a
    failed run completion while an exception is already leaving the
    ``with`` body: that exception is the one the caller needs, so it is
    never replaced::

        # Silence them
        warnings.filterwarnings("ignore", category=TrackingFailureWarning)

        # Or turn them back into errors
        warnings.filterwarnings("error", category=TrackingFailureWarning)
    """





class TrackingProjectChangedWarning(UserWarning):
    """Warns that selecting a run also changed the active project.

    A run belongs to exactly one project, so adopting one from another
    project has to move the client with it.  That is not a failure, and
    it is not what the caller literally asked for either, so it is said
    out loud rather than done quietly — a later ``log_metric`` would
    otherwise land somewhere the caller did not choose.

    Its own category, so it can be silenced by a script that switches
    deliberately without also silencing
    :class:`TrackingFailureWarning`::

        warnings.filterwarnings(
            "ignore", category=TrackingProjectChangedWarning
        )
    """





class TrackingUsageError(RuntimeError):
    """Raised when a tracking call itself is malformed.

    Safe mode (``safe=True``, :meth:`~embedl_hub.tracking.Client.safe_log`)
    tolerates an unreliable Hub, not a malformed call, so this is raised
    even inside a safe scope: a warning the caller cannot act on would
    just hide the mistake.

    It is marked with its own class rather than inferred from the
    exception type, because the type does not carry the distinction — a
    missing project or absent active run raises :class:`RuntimeError`
    too, and those *are* degradable, being the downstream consequence of
    a failure the caller already chose to tolerate.

    Subclasses :class:`RuntimeError` so that existing handlers and the
    ``:raises RuntimeError:`` contracts keep working.
    """



class StorageQuotaExceededError(DomainError):
    """Raised when storage quota is exceeded."""

    def __init__(self, file_path: Path, message: str | None = None) -> None:
        self.file_path = file_path
        if message is None:
            message = "Storage quota exceeded. Please delete some artifacts or contact support."
        super().__init__(message)


class FileTooLargeError(DomainError):
    """Raised when the file for an artifact is too large to upload."""

    def __init__(self, file_path: Path, message: str | None = None) -> None:
        self.file_path = file_path
        if message is None:
            message = "File too large to upload. Please try a smaller file."
        super().__init__(message)


class ArtifactUploadError(DomainError):
    """Raised when the file upload for an artifact fails."""

    def __init__(self, file_path: Path, message: str | None = None) -> None:
        self.file_path = file_path
        if message is None:
            message = "File upload failed. Please try again."
        super().__init__(message)


class UnsupportedDeviceError(DomainError):
    """Raised when an unsupported device is specified."""

    def __init__(self, device: str, message: str | None = None) -> None:
        self.device = device
        if message is None:
            message = f"The device '{device}' is not supported."
        super().__init__(message)


class JobQuotaExceededError(DomainError):
    """Raised when the job quota is exceeded."""

    def __init__(self, message: str | None = None) -> None:
        if message is None:
            message = "Job quota exceeded. Please try again later or contact support."
        super().__init__(message)


class InvalidRunLineageError(DomainError):
    """Raised when the API rejects an invalid parent run link."""

    def __init__(self, message: str | None = None) -> None:
        if message is None:
            message = "Invalid run lineage."
        super().__init__(message)


def raise_if_artifact_error(api_error: ApiError, *, file_path: Path):
    """Re-raise a domain-specific exception if the API error is related to artifacts."""

    for error in api_error.errors:
        if error.code == ErrorCode.FILE_TOO_LARGE.value:
            raise FileTooLargeError(
                file_path, message=error.detail
            ) from api_error

        if error.code == ErrorCode.STORAGE_QUOTA_EXCEEDED.value:
            raise StorageQuotaExceededError(
                file_path, message=error.detail
            ) from api_error


def raise_if_job_error(api_error: ApiError):
    """Re-raise a domain-specific exception if the API error is related to jobs."""

    for error in api_error.errors:
        if error.code == ErrorCode.JOB_QUOTA_EXCEEDED.value:
            raise JobQuotaExceededError(message=error.detail) from api_error


def raise_if_run_error(api_error: ApiError):
    """Re-raise a domain-specific exception if the API error is related to runs."""

    for error in api_error.errors:
        if error.code == ErrorCode.INVALID_RUN_LINEAGE.value:
            raise InvalidRunLineageError(
                message=error.detail or error.title
            ) from api_error

        text = " ".join(
            part for part in [error.title, error.detail] if part
        ).lower()
        if (
            "parent run" in text
            or "lineage" in text
            or "compile runs cannot have a parent run" in text
        ):
            raise InvalidRunLineageError(
                message=error.detail or error.title
            ) from api_error