openai/chatkit-python
Publicmirrored fromhttps://github.com/openai/chatkit-pythonAvailable
chatkit/errors.py
61lines · modecode
| 1 | from abc import ABC |
| 2 | from enum import Enum |
| 3 | |
| 4 | |
| 5 | # Not a closed enum, new error codes can and will be added as needed |
| 6 | class ErrorCode(str, Enum): |
| 7 | STREAM_ERROR = "stream.error" |
| 8 | |
| 9 | |
| 10 | DEFAULT_STATUS: dict[ErrorCode, int] = { |
| 11 | ErrorCode.STREAM_ERROR: 500, |
| 12 | } |
| 13 | |
| 14 | DEFAULT_ALLOW_RETRY: dict[ErrorCode, bool] = { |
| 15 | ErrorCode.STREAM_ERROR: True, |
| 16 | } |
| 17 | |
| 18 | |
| 19 | class BaseStreamError(ABC, Exception): |
| 20 | allow_retry: bool |
| 21 | |
| 22 | |
| 23 | class StreamError(BaseStreamError): |
| 24 | """ |
| 25 | Error with a specific error code that maps to a localized user-facing |
| 26 | error message. |
| 27 | """ |
| 28 | |
| 29 | code: ErrorCode |
| 30 | |
| 31 | def __init__( |
| 32 | self, |
| 33 | code: ErrorCode, |
| 34 | *, |
| 35 | allow_retry: bool | None = None, |
| 36 | ): |
| 37 | self.code = code |
| 38 | self.status_code = DEFAULT_STATUS.get(code, 500) |
| 39 | if allow_retry is None: |
| 40 | self.allow_retry = DEFAULT_ALLOW_RETRY.get(code, False) |
| 41 | else: |
| 42 | self.allow_retry = allow_retry |
| 43 | |
| 44 | |
| 45 | class CustomStreamError(BaseStreamError): |
| 46 | """ |
| 47 | Error with a custom user-facing error message. The message should be |
| 48 | localized as needed before raising the error. |
| 49 | """ |
| 50 | |
| 51 | message: str |
| 52 | """The user-facing error message to display.""" |
| 53 | |
| 54 | def __init__( |
| 55 | self, |
| 56 | message: str, |
| 57 | *, |
| 58 | allow_retry: bool = False, |
| 59 | ): |
| 60 | self.message = message |
| 61 | self.allow_retry = allow_retry |