feat: Add custom exceptions and replace direct Exception usage (#19)

- Defined custom exception classes for the project
- Replaced direct usage of Exception with custom exceptions
- Improved error handling and code clarity with specific exception types

cozepy/__init__.py

@@ -24,6 +24,7 @@
 from .config import COZE_COM_BASE_URL, COZE_CN_BASE_URL
 from .conversations import Conversation
 from .coze import Coze
+from .exception import CozeError, CozeAPIError, CozeEventError
 from .files import File
 from .model import (
     TokenPaged,
@@ -92,6 +93,10 @@
     "COZE_CN_BASE_URL",
     # coze
     "Coze",
+    # exception
+    "CozeError",
+    "CozeAPIError",
+    "CozeEventError",
     # model
     "TokenPaged",
     "NumberPaged",

cozepy/chat/__init__.py

@@ -3,6 +3,7 @@
 from typing import Dict, List, Iterator, Union, TYPE_CHECKING, Optional
 
 from cozepy.auth import Auth
+from cozepy.exception import CozeEventError
 from cozepy.model import CozeModel
 from cozepy.request import Requester
 
@@ -248,21 +249,21 @@ def __next__(self) -> ChatEvent:
                 if event == "":
                     event = line[6:]
                 else:
-                    raise Exception(f"invalid event: {line}")
+                    raise CozeEventError("event", line)
             elif line.startswith("data:"):
                 if data == "":
                     data = line[5:]
                 else:
-                    raise Exception(f"invalid event: {line}")
+                    raise CozeEventError("data", line)
             else:
-                raise Exception(f"invalid event: {line}")
+                raise CozeEventError("", line)
 
             times += 1
 
         if event == ChatEventType.done:
             raise StopIteration
         elif event == ChatEventType.error:
-            raise Exception(f"error event: {line}")
+            raise Exception(f"error event: {line}")  # TODO: error struct format
         elif event in [
             ChatEventType.conversation_message_delta,
             ChatEventType.conversation_message_completed,
@@ -279,7 +280,7 @@ def __next__(self) -> ChatEvent:
         ]:
             return ChatEvent(event=event, chat=Chat.model_validate(json.loads(data)))
         else:
-            raise Exception(f"unknown event: {event}, data: {data}")
+            raise ValueError(f"invalid chat.event: {event}, {data}")
 
 
 class ChatClient(object):

cozepy/exception.py

@@ -0,0 +1,35 @@
+class CozeError(Exception):
+    """
+    base class for all coze errors
+    """
+
+    pass
+
+
+class CozeAPIError(CozeError):
+    """
+    base class for all api errors
+    """
+
+    def __init__(self, code: int = None, msg: str = "", logid: str = None):
+        self.code = code
+        self.msg = msg
+        self.logid = logid
+        if code and code > 0:
+            super().__init__(f"code: {code}, msg: {msg}, logid: {logid}")
+        else:
+            super().__init__(f"msg: {msg}, logid: {logid}")
+
+
+class CozeEventError(CozeError):
+    """
+    base class for all event errors
+    """
+
+    def __init__(self, field: str = "", data: str = ""):
+        self.field = field
+        self.data = data
+        if field:
+            super().__init__(f"invalid event, field: {field}, data: {data}")
+        else:
+            super().__init__(f"invalid event, data: {data}")

cozepy/request.py

@@ -1,22 +1,14 @@
-from typing import (
-    TYPE_CHECKING,
-    Tuple,
-    Optional,
-    Union,
-    List,
-    Iterator,
-)
-from typing_extensions import get_args, get_origin  # compatibility with python 3.7
-
+from typing import TYPE_CHECKING, Tuple, Optional, Union, List, Iterator, Type, TypeVar
+from pydantic import BaseModel
 import httpx
 from httpx import Response
+from typing_extensions import get_args, get_origin  # compatibility with python 3.7
+
+from cozepy.exception import CozeAPIError
 
 if TYPE_CHECKING:
     from cozepy.auth import Auth
 
-from typing import Type, TypeVar
-
-from pydantic import BaseModel
 
 T = TypeVar("T", bound=BaseModel)
 
@@ -63,18 +55,16 @@ def request(
             json=body,
             files=files,
         )
+        logid = r.headers.get("x-tt-logid")
         if stream:
             return r.iter_lines()
 
         code, msg, data = self._parse_requests_code_msg(r, data_field)
 
         if code is not None and code > 0:
-            # TODO: Exception 自定义类型
-            logid = r.headers.get("x-tt-logid")
-            raise Exception(f"{code}: {msg}, logid:{logid}")
+            raise CozeAPIError(code, msg, logid)
         elif code is None and msg != "":
-            logid = r.headers.get("x-tt-logid")
-            raise Exception(f"{msg}, logid:{logid}")
+            raise CozeAPIError(code, msg, logid)
         if get_origin(model) is list:
             item_model = get_args(model)[0]
             return [item_model.model_validate(item) for item in data]

cozepy/workflows/runs/__init__.py

@@ -1,6 +1,7 @@
 from enum import Enum
 from typing import Dict, Any, Iterator
 
+from cozepy.exception import CozeEventError
 from cozepy.auth import Auth
 from cozepy.model import CozeModel
 from cozepy.request import Requester
@@ -120,19 +121,19 @@ def __next__(self) -> WorkflowEvent:
                 if event == "":
                     id = line[3:]
                 else:
-                    raise Exception(f"invalid event: {line}")
+                    raise CozeEventError("id", line)
             elif line.startswith("event:"):
                 if event == "":
                     event = line[6:].strip()
                 else:
-                    raise Exception(f"invalid event: {line}")
+                    raise CozeEventError("event", line)
             elif line.startswith("data:"):
                 if data == "":
                     data = line[5:]
                 else:
-                    raise Exception(f"invalid event: {line}")
+                    raise CozeEventError("data", line)
             else:
-                raise Exception(f"invalid event: {line}")
+                raise CozeEventError("", line)
 
             times += 1
 
@@ -155,7 +156,7 @@ def __next__(self) -> WorkflowEvent:
                 interrupt=WorkflowEventInterrupt.model_validate_json(data),
             )
         else:
-            raise Exception(f"unknown event: {line}")
+            raise ValueError(f"invalid workflows.event: {event}, {data}")
 
 
 class WorkflowsClient(object):

tests/test_exception.py

@@ -0,0 +1,15 @@
+from cozepy import CozeAPIError
+
+
+def test_coze_api_error():
+    err = CozeAPIError(1, "msg", "logid")
+    assert err.code == 1
+    assert err.msg == "msg"
+    assert err.logid == "logid"
+    assert str(err) == "code: 1, msg: msg, logid: logid"
+
+    err = CozeAPIError(None, "msg", "logid")
+    assert err.code is None
+    assert err.msg == "msg"
+    assert err.logid == "logid"
+    assert str(err) == "msg: msg, logid: logid"