File size: 2,251 Bytes
c481f8a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 | 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)
|