diff --git a/cozepy/__init__.py b/cozepy/__init__.py index b7f390a..925b5c6 100644 --- a/cozepy/__init__.py +++ b/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", ] diff --git a/cozepy/auth.py b/cozepy/auth.py index 7386a58..9781f43 100644 --- a/cozepy/auth.py +++ b/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,7 +40,7 @@ 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): @@ -48,12 +48,12 @@ 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): @@ -61,28 +61,26 @@ 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): diff --git a/cozepy/bot.py b/cozepy/bot.py index c927dc6..0f239e2 100644 --- a/cozepy/bot.py +++ b/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, diff --git a/cozepy/config.py b/cozepy/config.py index 487230a..d2cd7d4 100644 --- a/cozepy/config.py +++ b/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" diff --git a/cozepy/coze.py b/cozepy/coze.py index 5a9d088..b04543e 100644 --- a/cozepy/coze.py +++ b/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 diff --git a/cozepy/model.py b/cozepy/model.py index e02f905..351e068 100644 --- a/cozepy/model.py +++ b/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})" + ) diff --git a/cozepy/request.py b/cozepy/request.py index 8b4ffe2..5bf051a 100644 --- a/cozepy/request.py +++ b/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 diff --git a/tests/test_auth.py b/tests/test_auth.py index e678421..e63e9c7 100644 --- a/tests/test_auth.py +++ b/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 == "" diff --git a/tests/test_bot.py b/tests/test_bot.py index e0fb235..333ce3a 100644 --- a/tests/test_bot.py +++ b/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