format code

cozepy/__init__.py

@@ -4,15 +4,12 @@
 from .model import TokenPaged, NumberPaged
 
 __all__ = [
-    'ApplicationOAuth',
-    'Auth',
-    'TokenAuth',
-
-    'COZE_COM_BASE_URL',
-    'COZE_CN_BASE_URL',
-
-    'Coze',
-
-    'TokenPaged',
-    'NumberPaged'
+    "ApplicationOAuth",
+    "Auth",
+    "TokenAuth",
+    "COZE_COM_BASE_URL",
+    "COZE_CN_BASE_URL",
+    "Coze",
+    "TokenPaged",
+    "NumberPaged",
 ]

cozepy/auth.py

@@ -11,8 +11,8 @@
 
 
 def _random_hex(length):
-    hex_characters = '0123456789abcdef'
-    return ''.join(random.choice(hex_characters) for _ in range(length))
+    hex_characters = "0123456789abcdef"
+    return "".join(random.choice(hex_characters) for _ in range(length))
 
 
 class OAuthToken(CozeModel):
@@ -21,9 +21,9 @@ class OAuthToken(CozeModel):
     # How long the access token is valid, in seconds (UNIX timestamp)
     expires_in: int
     # An OAuth 2.0 refresh token. The app can use this token to acquire other access tokens after the current access token expires. Refresh tokens are long-lived.
-    refresh_token: str = ''
+    refresh_token: str = ""
     # fixed: Bearer
-    token_type: str = ''
+    token_type: str = ""
 
 
 class DeviceAuthCode(CozeModel):
@@ -40,49 +40,47 @@ class DeviceAuthCode(CozeModel):
 
     @property
     def verification_url(self):
-        return f'{self.verification_uri}?user_code={self.user_code}'
+        return f"{self.verification_uri}?user_code={self.user_code}"
 
 
 class ApplicationOAuth(object):
     """
     App OAuth process to support obtaining token and refreshing token.
     """
 
-    def __init__(self, client_id: str, client_secret: str = '', base_url: str = COZE_COM_BASE_URL):
+    def __init__(self, client_id: str, client_secret: str = "", base_url: str = COZE_COM_BASE_URL):
         self._client_id = client_id
         self._client_secret = client_secret
         self._base_url = base_url
         self._api_endpoint = urlparse(base_url).netloc
-        self._token = ''
+        self._token = ""
         self._requester = Requester()
 
     def jwt_auth(self, private_key: str, kid: str, ttl: int):
         """
         Get the token by jwt with jwt auth flow.
         """
         jwt_token = self._gen_jwt(self._api_endpoint, private_key, self._client_id, kid, 3600)
-        url = f'{self._base_url}/api/permission/oauth2/token'
-        headers = {
-            'Authorization': f'Bearer {jwt_token}'
-        }
+        url = f"{self._base_url}/api/permission/oauth2/token"
+        headers = {"Authorization": f"Bearer {jwt_token}"}
         body = {
-            'duration_seconds': ttl,
-            'grant_type': 'urn:ietf:params:oauth:grant-type:jwt-bearer',
+            "duration_seconds": ttl,
+            "grant_type": "urn:ietf:params:oauth:grant-type:jwt-bearer",
         }
-        return self._requester.request('post', url, OAuthToken, headers=headers, body=body)
+        return self._requester.request("post", url, OAuthToken, headers=headers, body=body)
 
     def _gen_jwt(self, api_endpoint: str, private_key: str, client_id: str, kid: str, ttl: int):
         now = int(time.time())
-        header = {'alg': 'RS256', 'typ': 'JWT', 'kid': kid}
+        header = {"alg": "RS256", "typ": "JWT", "kid": kid}
         payload = {
             "iss": client_id,
-            'aud': api_endpoint,
+            "aud": api_endpoint,
             "iat": now,
             "exp": now + ttl,
-            'jti': _random_hex(16),
+            "jti": _random_hex(16),
         }
         s = jwt.encode(header, payload, private_key)
-        return s.decode('utf-8')
+        return s.decode("utf-8")
 
 
 class Auth(abc.ABC):

cozepy/bot.py

@@ -13,7 +13,7 @@ class BotPromptInfo(CozeModel):
 
 class BotOnboardingInfo(CozeModel):
     # Configured prologue of the bot.
-    prologue: str = ''
+    prologue: str = ""
     # The list of recommended questions configured for the bot.
     # This field will not be returned when user suggested questions are not enabled.
     suggested_questions: List[str] = []
@@ -113,30 +113,27 @@ def get_online_info_v1(self, *, bot_id: str) -> Bot:
         :docs: https://www.coze.com/docs/developer_guides/get_metadata?_lang=en
         :calls: `GET /v1/bot/get_online_info`
         """
-        url = f'{self._base_url}/v1/bot/get_online_info'
-        params = {
-            'bot_id': bot_id
-        }
+        url = f"{self._base_url}/v1/bot/get_online_info"
+        params = {"bot_id": bot_id}
 
-        return self._requester.request('get', url, Bot, params=params)
+        return self._requester.request("get", url, Bot, params=params)
 
-    def list_published_bots_v1(self, *,
-                               space_id: str,
-                               page_num: int = 1,
-                               page_size: int = 20) -> NumberPaged[SimpleBot]:
+    def list_published_bots_v1(
+        self, *, space_id: str, page_num: int = 1, page_size: int = 20
+    ) -> NumberPaged[SimpleBot]:
         """
         Get the bots published as API service.
 
         :docs: https://www.coze.com/docs/developer_guides/published_bots_list?_lang=en
         :calls: `GET /v1/space/published_bots_list`
         """
-        url = f'{self._base_url}/v1/space/published_bots_list'
+        url = f"{self._base_url}/v1/space/published_bots_list"
         params = {
-            'space_id': space_id,
-            'page_size': page_size,
-            'page_index': page_num,
+            "space_id": space_id,
+            "page_size": page_size,
+            "page_index": page_num,
         }
-        data = self._requester.request('get', url, self._PrivateListPublishedBotsV1Data, params=params)
+        data = self._requester.request("get", url, self._PrivateListPublishedBotsV1Data, params=params)
         return NumberPaged(
             items=data.space_bots,
             page_num=page_num,

cozepy/config.py

@@ -1,2 +1,2 @@
-COZE_COM_BASE_URL = 'https://api.coze.com'
-COZE_CN_BASE_URL = 'https://api.coze.cn'
+COZE_COM_BASE_URL = "https://api.coze.com"
+COZE_CN_BASE_URL = "https://api.coze.cn"

cozepy/coze.py

@@ -9,10 +9,11 @@
 
 
 class Coze(object):
-    def __init__(self,
-                 auth: Auth,
-                 base_url: str = COZE_COM_BASE_URL,
-                 ):
+    def __init__(
+        self,
+        auth: Auth,
+        base_url: str = COZE_COM_BASE_URL,
+    ):
         self._auth = auth
         self._base_url = base_url
         self._requester = Requester(auth=auth)
@@ -21,8 +22,9 @@ def __init__(self,
         self._bot = None
 
     @property
-    def bot(self) -> 'BotClient':
+    def bot(self) -> "BotClient":
         if not self._bot:
             from cozepy.bot import BotClient
+
             self._bot = BotClient(self._base_url, self._auth, self._requester)
         return self._bot

cozepy/model.py

@@ -6,9 +6,7 @@
 
 
 class CozeModel(BaseModel):
-    model_config = ConfigDict(
-        protected_namespaces=()
-    )
+    model_config = ConfigDict(protected_namespaces=())
 
 
 class PagedBase(Generic[T]):
@@ -30,8 +28,8 @@ class TokenPaged(PagedBase[T]):
     return is next_page_token + has_more.
     """
 
-    def __init__(self, items: List[T], next_page_token: str = '', has_more: bool = None):
-        has_more = has_more if has_more is not None else next_page_token != ''
+    def __init__(self, items: List[T], next_page_token: str = "", has_more: bool = None):
+        has_more = has_more if has_more is not None else next_page_token != ""
         super().__init__(items, has_more)
         self.next_page_token = next_page_token
 
@@ -48,4 +46,6 @@ def __init__(self, items: List[T], page_num: int, page_size: int, total: int = N
         self.total = total
 
     def __repr__(self):
-        return f"NumberPaged(items={self.items}, page_num={self.page_num}, page_size={self.page_size}, total={self.total})"
+        return (
+            f"NumberPaged(items={self.items}, page_num={self.page_num}, page_size={self.page_size}, total={self.total})"
+        )

cozepy/request.py

@@ -10,7 +10,7 @@
 
 from pydantic import BaseModel
 
-T = TypeVar('T', bound=BaseModel)
+T = TypeVar("T", bound=BaseModel)
 
 
 def json_obj_to_pydantic(json_obj: dict, model: Type[T]) -> T:
@@ -25,13 +25,18 @@ class Requester(object):
     http request helper class.
     """
 
-    def __init__(self,
-                 auth: 'Auth' = None
-                 ):
+    def __init__(self, auth: "Auth" = None):
         self._auth = auth
 
-    def request(self, method: str, url: str, model: Type[T], params: dict = None, headers: dict = None,
-                body: dict = None, ) -> T:
+    def request(
+        self,
+        method: str,
+        url: str,
+        model: Type[T],
+        params: dict = None,
+        headers: dict = None,
+        body: dict = None,
+    ) -> T:
         """
         Send a request to the server.
         """
@@ -45,11 +50,11 @@ def request(self, method: str, url: str, model: Type[T], params: dict = None, he
 
         if code is not None and code > 0:
             # TODO: Exception 自定义类型
-            logid = r.headers.get('x-tt-logid')
-            raise Exception(f'{code}: {msg}, logid:{logid}')
+            logid = r.headers.get("x-tt-logid")
+            raise Exception(f"{code}: {msg}, logid:{logid}")
         elif code is None and msg != "":
-            logid = r.headers.get('x-tt-logid')
-            raise Exception(f'{msg}, logid:{logid}')
+            logid = r.headers.get("x-tt-logid")
+            raise Exception(f"{msg}, logid:{logid}")
         return model.model_validate(data)
 
     async def arequest(self, method: str, path: str, **kwargs) -> dict:
@@ -65,10 +70,10 @@ def __parse_requests_code_msg(self, r: Response) -> Tuple[Optional[int], str, Op
             r.raise_for_status()
             return
 
-        if 'code' in json and 'msg' in json and int(json['code']) > 0:
-            return int(json['code']), json['msg'], json['data']
-        if 'error_message' in json and json['error_message'] != '':
-            return None, json['error_message'], None
-        if 'data' in json:
-            return 0, '', json['data']
-        return 0, '', json
+        if "code" in json and "msg" in json and int(json["code"]) > 0:
+            return int(json["code"]), json["msg"], json["data"]
+        if "error_message" in json and json["error_message"] != "":
+            return None, json["error_message"], None
+        if "data" in json:
+            return 0, "", json["data"]
+        return 0, "", json

tests/test_auth.py

@@ -5,16 +5,16 @@
 
 
 def test_jwt_auth():
-    client_id = os.getenv('COZE_JWT_AUTH_CLIENT_ID')
-    private_key = os.getenv('COZE_JWT_AUTH_PRIVATE_KEY')
-    key_id = os.getenv('COZE_JWT_AUTH_KEY_ID')
+    client_id = os.getenv("COZE_JWT_AUTH_CLIENT_ID")
+    private_key = os.getenv("COZE_JWT_AUTH_PRIVATE_KEY")
+    key_id = os.getenv("COZE_JWT_AUTH_KEY_ID")
 
     app = ApplicationOAuth(
         client_id,
         base_url=COZE_CN_BASE_URL,
     )
     token = app.jwt_auth(private_key, key_id, 30)
-    assert token.access_token != ''
-    assert token.token_type == 'Bearer'
+    assert token.access_token != ""
+    assert token.token_type == "Bearer"
     assert token.expires_in - int(time.time()) <= 30
-    assert token.refresh_token == ''
+    assert token.refresh_token == ""

tests/test_bot.py

@@ -6,12 +6,12 @@
 
 class TestBotClient(TestCase):
     def test_list_published_bots_v1(self):
-        space_id = os.getenv('SPACE_ID_1').strip()
-        token = os.getenv('COZE_TOKEN').strip()
+        space_id = os.getenv("SPACE_ID_1").strip()
+        token = os.getenv("COZE_TOKEN").strip()
         for i in token:
-            print('token', i)
+            print("token", i)
         auth = TokenAuth(token)
-        cli = Coze(auth=auth, base_url='https://api.coze.cn')
+        cli = Coze(auth=auth, base_url="https://api.coze.cn")
 
         res = cli.bot.list_published_bots_v1(space_id=space_id, page_size=2)
         assert res.total > 1