service/errors.py

from __future__ import annotations

from dataclasses import dataclass
from typing import Any, Dict


ERROR_CODE_SUCCESS = 200

ERROR_CODE_AUTH = 10001
ERROR_CODE_RATE = 10002
ERROR_CODE_RISK = 10003
ERROR_CODE_PARSE = 10004
ERROR_CODE_TIMEOUT = 10005
ERROR_CODE_MISSING_DEPENDENCY = 10006
ERROR_CODE_INVALID_TARGET = 10007
ERROR_CODE_TASK_NOT_FOUND = 10008
ERROR_CODE_TASK_NOT_READY = 10009
ERROR_CODE_ORCHESTRATOR_DB = 10010


@dataclass(frozen=True)
class ApiErrorSpec:
    code: int
    http_status: int
    default_msg: str


ERROR_SPECS: Dict[str, ApiErrorSpec] = {
    "auth": ApiErrorSpec(code=ERROR_CODE_AUTH, http_status=401, default_msg="auth required"),
    "rate": ApiErrorSpec(code=ERROR_CODE_RATE, http_status=429, default_msg="rate limited"),
    "risk": ApiErrorSpec(code=ERROR_CODE_RISK, http_status=403, default_msg="risk control"),
    "parse": ApiErrorSpec(code=ERROR_CODE_PARSE, http_status=500, default_msg="parse error"),
    "timeout": ApiErrorSpec(code=ERROR_CODE_TIMEOUT, http_status=504, default_msg="timeout"),
    "missing_dependency": ApiErrorSpec(
        code=ERROR_CODE_MISSING_DEPENDENCY,
        http_status=500,
        default_msg="missing dependency",
    ),
    "invalid_target": ApiErrorSpec(
        code=ERROR_CODE_INVALID_TARGET,
        http_status=400,
        default_msg="invalid target",
    ),
    "task_not_found": ApiErrorSpec(
        code=ERROR_CODE_TASK_NOT_FOUND,
        http_status=404,
        default_msg="task not found",
    ),
    "task_not_ready": ApiErrorSpec(
        code=ERROR_CODE_TASK_NOT_READY,
        http_status=409,
        default_msg="task not ready",
    ),
    "orchestrator_db": ApiErrorSpec(
        code=ERROR_CODE_ORCHESTRATOR_DB,
        http_status=503,
        default_msg="orchestrator db unavailable",
    ),
}


class ApiException(Exception):
    def __init__(self, kind: str, msg: str | None = None, data: Any = None):
        self.kind = (kind or "").strip() or "parse"
        spec = ERROR_SPECS.get(self.kind) or ERROR_SPECS["parse"]
        self.code = spec.code
        self.http_status = spec.http_status
        self.msg = msg if msg is not None and str(msg).strip() != "" else spec.default_msg
        self.data = data
        super().__init__(self.msg)