app/generation/domain/errors.py

from __future__ import annotations

from app.core.exceptions import MediaAPIError
from app.generation.domain.enums import WorkerErrorCategory


class GenerationError(MediaAPIError):
    code = "GENERATION_ERROR"
    status_code = 400


class GenerationProviderUnavailableError(GenerationError):
    code = "GENERATION_PROVIDER_UNAVAILABLE"
    status_code = 503


class GenerationCapabilityUnsupportedError(GenerationError):
    code = "GENERATION_CAPABILITY_UNSUPPORTED"
    status_code = 422


class GenerationValidationError(GenerationError):
    code = "GENERATION_VALIDATION_ERROR"
    status_code = 422


class GenerationRequestNotFoundError(GenerationError):
    code = "GENERATION_REQUEST_NOT_FOUND"
    status_code = 404


class GenerationJobNotFoundError(GenerationError):
    code = "GENERATION_JOB_NOT_FOUND"
    status_code = 404


class GenerationIdempotencyConflictError(GenerationError):
    code = "GENERATION_IDEMPOTENCY_CONFLICT"
    status_code = 409


class GenerationTransitionError(GenerationError):
    code = "GENERATION_INVALID_STATE_TRANSITION"
    status_code = 409


class GenerationInputAssetNotFoundError(GenerationError):
    code = "GENERATION_INPUT_ASSET_NOT_FOUND"
    status_code = 404


class GenerationProviderJobConflictError(GenerationError):
    """A provider job ID is already bound to a different logical job."""

    code = "GENERATION_PROVIDER_JOB_CONFLICT"
    status_code = 409


class GenerationOutputConflictError(GenerationError):
    code = "GENERATION_OUTPUT_CONFLICT"
    status_code = 409


class GenerationWorkerError(GenerationError):
    """Safe normalised failure returned by a remote generation worker."""

    code = "GENERATION_WORKER_ERROR"
    status_code = 502

    _CLIENT_STATUS_BY_CATEGORY = {
        WorkerErrorCategory.INVALID_REQUEST: 422,
        WorkerErrorCategory.WORKER_UNAVAILABLE: 503,
        WorkerErrorCategory.WORKER_NOT_READY: 503,
        WorkerErrorCategory.TIMEOUT: 504,
        WorkerErrorCategory.RATE_LIMITED: 503,
    }

    def __init__(
        self,
        *,
        category: WorkerErrorCategory,
        message: str,
        retryable: bool,
        http_status: int | None = None,
    ) -> None:
        super().__init__(message)
        self.category = category
        self.retryable = retryable
        self.http_status = http_status
        self.status_code = self._CLIENT_STATUS_BY_CATEGORY.get(category, 502)


class GenerationCancellationError(GenerationWorkerError):
    code = "GENERATION_CANCELLATION_FAILED"

    def __init__(self, message: str = "Generation cancellation could not be confirmed.") -> None:
        super().__init__(
            category=WorkerErrorCategory.CANCELLATION_ERROR,
            message=message,
            retryable=False,
        )


class GenerationOutputError(GenerationWorkerError):
    code = "GENERATION_OUTPUT_INVALID"

    def __init__(self, message: str = "Generation worker output is invalid.") -> None:
        super().__init__(
            category=WorkerErrorCategory.OUTPUT_ERROR,
            message=message,
            retryable=False,
        )