feat: Add v1/users/me API endpoint to retrieve current authorized use…

…r info (#177)

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

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:

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)

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)

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