122 lines
3.2 KiB
Python
122 lines
3.2 KiB
Python
from __future__ import annotations
|
||
|
||
from open_chatcaht.utils import is_dict
|
||
from typing import Any, Optional, cast
|
||
from typing_extensions import Literal
|
||
|
||
import httpx
|
||
|
||
__all__ = [
|
||
"BadRequestError",
|
||
"AuthenticationError",
|
||
"PermissionDeniedError",
|
||
"NotFoundError",
|
||
"ConflictError",
|
||
"UnprocessableEntityError",
|
||
"RateLimitError",
|
||
"InternalServerError",
|
||
]
|
||
|
||
|
||
class ChatChatError(Exception):
|
||
pass
|
||
|
||
|
||
class APIError(ChatChatError):
|
||
message: str
|
||
request: httpx.Request
|
||
|
||
body: object | None
|
||
"""
|
||
API响应体。
|
||
如果API响应了一个有效的JSON结构，那么这个属性将是
|
||
解码结果。
|
||
如果它不是一个有效的JSON结构，那么这将是原始响应。
|
||
如果没有与此错误相关的响应，那么它将是' None '。
|
||
"""
|
||
|
||
code: Optional[str] = None
|
||
param: Optional[str] = None
|
||
type: Optional[str]
|
||
|
||
def __init__(self, message: str, request: httpx.Request, *, body: object | None) -> None:
|
||
super().__init__(message)
|
||
self.request = request
|
||
self.message = message
|
||
self.body = body
|
||
|
||
if is_dict(body):
|
||
self.code = cast(str, body.get("code"))
|
||
self.param = cast(str, body.get("param"))
|
||
self.type = cast(str, body.get("type"))
|
||
else:
|
||
self.code = None
|
||
self.param = None
|
||
self.type = None
|
||
|
||
|
||
class APIResponseValidationError(APIError):
|
||
response: httpx.Response
|
||
status_code: int
|
||
|
||
def __init__(self, response: httpx.Response, body: object | None, *, message: str | None = None) -> None:
|
||
super().__init__(message or "API返回的数据对预期的模式无效。", response.request, body=body)
|
||
self.response = response
|
||
self.status_code = response.status_code
|
||
|
||
|
||
class APIStatusError(APIError):
|
||
"""当API响应的状态码为4xx或5xx时引发。"""
|
||
|
||
response: httpx.Response
|
||
status_code: int
|
||
request_id: str | None
|
||
|
||
def __init__(self, message: str, *, response: httpx.Response, body: object | None) -> None:
|
||
super().__init__(message, response.request, body=body)
|
||
self.response = response
|
||
self.status_code = response.status_code
|
||
self.request_id = response.headers.get("x-request-id")
|
||
|
||
|
||
class APIConnectionError(APIError):
|
||
def __init__(self, *, message: str = "连接错误", request: httpx.Request) -> None:
|
||
super().__init__(message, request, body=None)
|
||
|
||
|
||
class APITimeoutError(APIConnectionError):
|
||
def __init__(self, request: httpx.Request) -> None:
|
||
super().__init__(message="请求超时", request=request)
|
||
|
||
|
||
class BadRequestError(APIStatusError):
|
||
status_code: Literal[400] = 400
|
||
|
||
|
||
class AuthenticationError(APIStatusError):
|
||
status_code: Literal[401] = 401
|
||
|
||
|
||
class PermissionDeniedError(APIStatusError):
|
||
status_code: Literal[403] = 403
|
||
|
||
|
||
class NotFoundError(APIStatusError):
|
||
status_code: Literal[404] = 404
|
||
|
||
|
||
class ConflictError(APIStatusError):
|
||
status_code: Literal[409] = 409
|
||
|
||
|
||
class UnprocessableEntityError(APIStatusError):
|
||
status_code: Literal[422] = 422
|
||
|
||
|
||
class RateLimitError(APIStatusError):
|
||
status_code: Literal[429] = 429
|
||
|
||
|
||
class InternalServerError(APIStatusError):
|
||
pass
|