1# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details.
  2
  3from __future__ import annotations
  4
  5from typing import TYPE_CHECKING, Any, Optional, cast
  6from typing_extensions import Literal
  7
  8import httpx
  9
 10from ._utils import is_dict
 11from ._models import construct_type
 12
 13if TYPE_CHECKING:
 14    from .types.chat import ChatCompletion
 15
 16__all__ = [
 17    "BadRequestError",
 18    "AuthenticationError",
 19    "PermissionDeniedError",
 20    "NotFoundError",
 21    "ConflictError",
 22    "UnprocessableEntityError",
 23    "RateLimitError",
 24    "InternalServerError",
 25    "LengthFinishReasonError",
 26    "ContentFilterFinishReasonError",
 27    "InvalidWebhookSignatureError",
 28]
 29
 30
 31class OpenAIError(Exception):
 32    pass
 33
 34
 35class APIError(OpenAIError):
 36    message: str
 37    request: httpx.Request
 38
 39    body: object | None
 40    """The API response body.
 41
 42    If the API responded with a valid JSON structure then this property will be the
 43    decoded result.
 44
 45    If it isn't a valid JSON structure then this will be the raw response.
 46
 47    If there was no response associated with this error then it will be `None`.
 48    """
 49
 50    code: Optional[str] = None
 51    param: Optional[str] = None
 52    type: Optional[str]
 53
 54    def __init__(self, message: str, request: httpx.Request, *, body: object | None) -> None:
 55        super().__init__(message)
 56        self.request = request
 57        self.message = message
 58        self.body = body
 59
 60        if is_dict(body):
 61            self.code = cast(Any, construct_type(type_=Optional[str], value=body.get("code")))
 62            self.param = cast(Any, construct_type(type_=Optional[str], value=body.get("param")))
 63            self.type = cast(Any, construct_type(type_=str, value=body.get("type")))
 64        else:
 65            self.code = None
 66            self.param = None
 67            self.type = None
 68
 69
 70class APIResponseValidationError(APIError):
 71    response: httpx.Response
 72    status_code: int
 73
 74    def __init__(self, response: httpx.Response, body: object | None, *, message: str | None = None) -> None:
 75        super().__init__(message or "Data returned by API invalid for expected schema.", response.request, body=body)
 76        self.response = response
 77        self.status_code = response.status_code
 78
 79
 80class APIStatusError(APIError):
 81    """Raised when an API response has a status code of 4xx or 5xx."""
 82
 83    response: httpx.Response
 84    status_code: int
 85    request_id: str | None
 86
 87    def __init__(self, message: str, *, response: httpx.Response, body: object | None) -> None:
 88        super().__init__(message, response.request, body=body)
 89        self.response = response
 90        self.status_code = response.status_code
 91        self.request_id = response.headers.get("x-request-id")
 92
 93
 94class APIConnectionError(APIError):
 95    def __init__(self, *, message: str = "Connection error.", request: httpx.Request) -> None:
 96        super().__init__(message, request, body=None)
 97
 98
 99class APITimeoutError(APIConnectionError):
100    def __init__(self, request: httpx.Request) -> None:
101        super().__init__(message="Request timed out.", request=request)
102
103
104class BadRequestError(APIStatusError):
105    status_code: Literal[400] = 400  # pyright: ignore[reportIncompatibleVariableOverride]
106
107
108class AuthenticationError(APIStatusError):
109    status_code: Literal[401] = 401  # pyright: ignore[reportIncompatibleVariableOverride]
110
111
112class PermissionDeniedError(APIStatusError):
113    status_code: Literal[403] = 403  # pyright: ignore[reportIncompatibleVariableOverride]
114
115
116class NotFoundError(APIStatusError):
117    status_code: Literal[404] = 404  # pyright: ignore[reportIncompatibleVariableOverride]
118
119
120class ConflictError(APIStatusError):
121    status_code: Literal[409] = 409  # pyright: ignore[reportIncompatibleVariableOverride]
122
123
124class UnprocessableEntityError(APIStatusError):
125    status_code: Literal[422] = 422  # pyright: ignore[reportIncompatibleVariableOverride]
126
127
128class RateLimitError(APIStatusError):
129    status_code: Literal[429] = 429  # pyright: ignore[reportIncompatibleVariableOverride]
130
131
132class InternalServerError(APIStatusError):
133    pass
134
135
136class LengthFinishReasonError(OpenAIError):
137    completion: ChatCompletion
138    """The completion that caused this error.
139
140    Note: this will *not* be a complete `ChatCompletion` object when streaming as `usage`
141          will not be included.
142    """
143
144    def __init__(self, *, completion: ChatCompletion) -> None:
145        msg = "Could not parse response content as the length limit was reached"
146        if completion.usage:
147            msg += f" - {completion.usage}"
148
149        super().__init__(msg)
150        self.completion = completion
151
152
153class ContentFilterFinishReasonError(OpenAIError):
154    def __init__(self) -> None:
155        super().__init__(
156            f"Could not parse response content as the request was rejected by the content filter",
157        )
158
159
160class InvalidWebhookSignatureError(ValueError):
161    """Raised when a webhook signature is invalid, meaning the computed signature does not match the expected signature."""