diff --git a/cozepy/__init__.py b/cozepy/__init__.py index a3784b9..a233655 100644 --- a/cozepy/__init__.py +++ b/cozepy/__init__.py @@ -91,6 +91,7 @@ ) from .request import AsyncHTTPClient, SyncHTTPClient from .templates import TemplateDuplicateResp, TemplateEntityType +from .users import User from .version import VERSION from .websockets.audio.speech import ( AsyncWebsocketsAudioSpeechClient, @@ -289,6 +290,8 @@ # templates "TemplateDuplicateResp", "TemplateEntityType", + # users + "User", # log "setup_logging", # config diff --git a/cozepy/coze.py b/cozepy/coze.py index 4535ef6..28f1a24 100644 --- a/cozepy/coze.py +++ b/cozepy/coze.py @@ -15,6 +15,7 @@ from .files import AsyncFilesClient, FilesClient from .knowledge import AsyncKnowledgeClient, KnowledgeClient # deprecated from .templates import AsyncTemplatesClient, TemplatesClient + from .users import AsyncUsersClient, UsersClient from .websockets import AsyncWebsocketsClient from .workflows import AsyncWorkflowsClient, WorkflowsClient from .workspaces import AsyncWorkspacesClient, WorkspacesClient @@ -42,6 +43,7 @@ def __init__( self._datasets: Optional[DatasetsClient] = None self._audio: Optional[AudioClient] = None self._templates: Optional[TemplatesClient] = None + self._users: Optional[UsersClient] = None @property def bots(self) -> "BotsClient": @@ -129,6 +131,14 @@ def templates(self) -> "TemplatesClient": self._templates = TemplatesClient(self._base_url, self._auth, self._requester) return self._templates + @property + def users(self) -> "UsersClient": + if not self._users: + from .users import UsersClient + + self._users = UsersClient(self._base_url, self._auth, self._requester) + return self._users + class AsyncCoze(object): def __init__( @@ -152,6 +162,7 @@ def __init__( self._workspaces: Optional[AsyncWorkspacesClient] = None self._audio: Optional[AsyncAudioClient] = None self._templates: Optional[AsyncTemplatesClient] = None + self._users: Optional[AsyncUsersClient] = None self._websockets: Optional[AsyncWebsocketsClient] = None @property @@ -240,6 +251,14 @@ def templates(self) -> "AsyncTemplatesClient": self._templates = AsyncTemplatesClient(self._base_url, self._auth, self._requester) return self._templates + @property + def users(self) -> "AsyncUsersClient": + if not self._users: + from .users import AsyncUsersClient + + self._users = AsyncUsersClient(self._base_url, self._auth, self._requester) + return self._users + @property def websockets(self) -> "AsyncWebsocketsClient": if not self._websockets: diff --git a/cozepy/users/__init__.py b/cozepy/users/__init__.py new file mode 100644 index 0000000..22928a8 --- /dev/null +++ b/cozepy/users/__init__.py @@ -0,0 +1,37 @@ +from typing import Optional + +from cozepy.auth import Auth +from cozepy.model import CozeModel +from cozepy.request import Requester +from cozepy.util import remove_url_trailing_slash + + +class User(CozeModel): + user_id: str + user_name: str + nick_name: str + avatar_url: str + + +class UsersClient(object): + def __init__(self, base_url: str, auth: Auth, requester: Requester): + self._base_url = remove_url_trailing_slash(base_url) + self._auth = auth + self._requester = requester + + def me(self, **kwargs) -> User: + url = f"{self._base_url}/v1/users/me" + headers: Optional[dict] = kwargs.get("headers") + return self._requester.request("get", url, False, User, headers=headers) + + +class AsyncUsersClient(object): + def __init__(self, base_url: str, auth: Auth, requester: Requester): + self._base_url = remove_url_trailing_slash(base_url) + self._auth = auth + self._requester = requester + + async def me(self, **kwargs) -> User: + url = f"{self._base_url}/v1/users/me" + headers: Optional[dict] = kwargs.get("headers") + return await self._requester.arequest("get", url, False, User, headers=headers) diff --git a/examples/users_me.py b/examples/users_me.py new file mode 100644 index 0000000..77ee7ea --- /dev/null +++ b/examples/users_me.py @@ -0,0 +1,53 @@ +import json +import logging +import os +from typing import Optional + +from cozepy import ( + COZE_CN_BASE_URL, + Coze, + DeviceOAuthApp, + TokenAuth, + setup_logging, +) + + +def get_coze_api_base() -> str: + # The default access is api.coze.com, but if you need to access api.coze.cn, + # please use base_url to configure the api endpoint to access + coze_api_base = os.getenv("COZE_API_BASE") + if coze_api_base: + return coze_api_base + + return COZE_CN_BASE_URL # default + + +def get_coze_api_token(workspace_id: Optional[str] = None) -> str: + # Get an access_token through personal access token or oauth. + coze_api_token = os.getenv("COZE_API_TOKEN") + if coze_api_token: + return coze_api_token + + coze_api_base = get_coze_api_base() + + device_oauth_app = DeviceOAuthApp(client_id="57294420732781205987760324720643.app.coze", base_url=coze_api_base) + device_code = device_oauth_app.get_device_code(workspace_id) + print(f"Please Open: {device_code.verification_url} to get the access token") + return device_oauth_app.get_access_token(device_code=device_code.device_code, poll=True).access_token + + +def setup_examples_logger(): + coze_log = os.getenv("COZE_LOG") + if coze_log: + setup_logging(logging.getLevelNamesMapping().get(coze_log.upper(), logging.INFO)) + + +setup_examples_logger() + +kwargs = json.loads(os.getenv("COZE_KWARGS") or "{}") + +if __name__ == "__main__": + coze = Coze(auth=TokenAuth(get_coze_api_token()), base_url=get_coze_api_base()) + + user = coze.users.me(**kwargs) + print(user) diff --git a/tests/test_users.py b/tests/test_users.py new file mode 100644 index 0000000..288fcf5 --- /dev/null +++ b/tests/test_users.py @@ -0,0 +1,51 @@ +import httpx +import pytest + +from cozepy import AsyncCoze, Coze, TokenAuth, User +from cozepy.util import random_hex +from tests.test_util import logid_key + + +def mock_retrieve_users_me( + respx_mock, +) -> User: + user = User( + user_id="user_id", + user_name=random_hex(10), + nick_name=random_hex(10), + avatar_url=random_hex(10), + ) + user._raw_response = httpx.Response( + 200, + json={"data": user.model_dump()}, + headers={logid_key(): random_hex(10)}, + ) + respx_mock.get("/v1/users/me").mock(user._raw_response) + return user + + +@pytest.mark.respx(base_url="https://api.coze.com") +class TestSyncUsers: + def test_sync_users_retrieve_me(self, respx_mock): + coze = Coze(auth=TokenAuth(token="token")) + + mock_user = mock_retrieve_users_me(respx_mock) + + user = coze.users.me() + assert user + assert user.response.logid == mock_user.response.logid + assert user.user_id == mock_user.user_id + + +@pytest.mark.respx(base_url="https://api.coze.com") +@pytest.mark.asyncio +class TestAsyncUsers: + async def test_async_users_retrieve_me(self, respx_mock): + coze = AsyncCoze(auth=TokenAuth(token="token")) + + mock_user = mock_retrieve_users_me(respx_mock) + + user = await coze.users.me() + assert user + assert user.response.logid == mock_user.response.logid + assert user.user_id == mock_user.user_id