feat: add upload and retrieve file api

cozepy/bot/v1/__init__.py

@@ -170,7 +170,7 @@ def retrieve(self, *, bot_id: str) -> Bot:
         Get the configuration information of the bot, which must have been published
         to the Bot as API channel.
 
-        :docs: https://www.coze.com/docs/developer_guides/get_metadata?_lang=en
+        :docs: https://www.coze.com/docs/developer_guides/get_metadata
         :calls: `GET /v1/bot/get_online_info`
         """
         url = f"{self._base_url}/v1/bot/get_online_info"
@@ -182,7 +182,7 @@ def list(self, *, space_id: str, page_num: int = 1, page_size: int = 20) -> Numb
         """
         Get the bots published as API service.
 
-        :docs: https://www.coze.com/docs/developer_guides/published_bots_list?_lang=en
+        :docs: https://www.coze.com/docs/developer_guides/published_bots_list
         :calls: `GET /v1/space/published_bots_list`
         """
         url = f"{self._base_url}/v1/space/published_bots_list"

cozepy/coze.py

@@ -9,6 +9,7 @@
     from .workspace import WorkspaceClient
     from .conversation import ConversationClient
     from .chat import ChatClient
+    from .file import FileClient
 
 
 class Coze(object):
@@ -26,6 +27,7 @@ def __init__(
         self._workspace = None
         self._conversation = None
         self._chat = None
+        self._file = None
 
     @property
     def bot(self) -> "BotClient":
@@ -58,3 +60,11 @@ def chat(self) -> "ChatClient":
 
             self._chat = ChatClient(self._base_url, self._auth, self._requester)
         return self._chat
+
+    @property
+    def file(self) -> "FileClient":
+        if not self._file:
+            from .file import FileClient
+
+            self._file = FileClient(self._base_url, self._auth, self._requester)
+        return self._file

cozepy/file/__init__.py

@@ -0,0 +1,23 @@
+from typing import TYPE_CHECKING
+
+from cozepy.auth import Auth
+from cozepy.request import Requester
+
+if TYPE_CHECKING:
+    from .v1 import FileClient as FileClientV1
+
+
+class FileClient(object):
+    def __init__(self, base_url: str, auth: Auth, requester: Requester):
+        self._base_url = base_url
+        self._auth = auth
+        self._requester = requester
+        self._v1 = None
+
+    @property
+    def v1(self) -> "FileClientV1":
+        if not self._v1:
+            from .v1 import FileClient
+
+            self._v1 = FileClient(self._base_url, self._auth, self._requester)
+        return self._v1

cozepy/file/v1/__init__.py

@@ -0,0 +1,72 @@
+from cozepy.auth import Auth
+from cozepy.model import CozeModel
+from cozepy.request import Requester
+
+
+class File(CozeModel):
+    # The ID of the uploaded file.
+    # 已上传的文件 ID。
+    id: str
+
+    # 文件的总字节数。
+    # The total byte size of the file.
+    bytes: int
+
+    # 文件的上传时间，格式为 10 位的 Unixtime 时间戳，单位为秒（s）。
+    # The upload time of the file, in the format of a 10-digit Unix timestamp in seconds (s).
+    created_at: int
+
+    # 文件名称。
+    # The name of the file.
+    file_name: str
+
+
+class FileClient(object):
+    def __init__(self, base_url: str, auth: Auth, requester: Requester):
+        self._base_url = base_url
+        self._auth = auth
+        self._requester = requester
+        self._v1 = None
+
+    def upload(self, file: str) -> File:
+        """
+        Upload files to Coze platform.
+
+        Local files cannot be used directly in messages. Before creating messages or conversations,
+        you need to call this interface to upload local files to the platform first.
+        After uploading the file, you can use it directly in multimodal content in messages
+        by specifying the file_id.
+
+        调用接口上传文件到扣子。
+
+        docs en: https://www.coze.com/docs/developer_guides/upload_files
+        docs zh: https://www.coze.cn/docs/developer_guides/upload_files
+
+        Args:
+            file: local file path
+
+        Returns:
+            File: file info
+        """
+        url = f"{self._base_url}/v1/files/upload"
+        files = {"file": open(file, "rb")}
+        return self._requester.request("get", url, File, files=files)
+
+    def retrieve(self, file_id: str):
+        """
+        Get the information of the specific file uploaded to Coze platform.
+
+        查看已上传的文件详情。
+
+        docs en: https://www.coze.com/docs/developer_guides/retrieve_files
+        docs cn: https://www.coze.cn/docs/developer_guides/retrieve_files
+
+        Args:
+            file_id: file id of file
+
+        Returns:
+            File: file info
+        """
+        url = f"{self._base_url}/v1/files/retrieve"
+        params = {"file_id": file_id}
+        return self._requester.request("get", url, File, params=params)

cozepy/request.py

@@ -1,6 +1,7 @@
+from typing import TYPE_CHECKING, Tuple, Optional, Union, List, get_origin, get_args, Iterator
+
 import requests
 from requests import Response
-from typing import TYPE_CHECKING, Tuple, Optional, Union, List, get_origin, get_args, Iterator
 
 if TYPE_CHECKING:
     from cozepy.auth import Auth
@@ -35,6 +36,7 @@ def request(
         params: dict = None,
         headers: dict = None,
         body: dict = None,
+        files: dict = None,
         stream: bool = False,
         data_field: str = "data",
     ) -> Union[T, List[T], Iterator[bytes]]:
@@ -45,7 +47,7 @@ def request(
             headers = {}
         if self._auth:
             self._auth.authentication(headers)
-        r = requests.request(method, url, params=params, headers=headers, json=body, stream=stream)
+        r = requests.request(method, url, params=params, headers=headers, json=body, files=files, stream=stream)
         if stream:
             return r.iter_lines()
 

tests/test_file.py

@@ -0,0 +1,16 @@
+from cozepy import Coze, COZE_CN_BASE_URL
+from tests.config import fixed_token_auth
+import unittest
+
+
+@unittest.skip("not available in not cn")
+def test_file_v1():
+    cli = Coze(auth=fixed_token_auth, base_url=COZE_CN_BASE_URL)
+
+    file = cli.file.v1.upload("./tests/test_file.py")
+    assert file is not None
+    assert file.id != ""
+
+    file_retrieve = cli.file.v1.retrieve(file_id=file.id)
+    assert file_retrieve is not None
+    assert file_retrieve.id == file.id