feat: Add v1/users/me API endpoint to retrieve current authorized user info #177
Conversation
WalkthroughThis pull request introduces user-related functionality to the Coze API library. It adds a new Changes
Sequence DiagramsequenceDiagram
participant Client as Coze Client
participant UsersClient as Users Client
participant API as Coze API
Client->>UsersClient: users.me()
UsersClient->>API: GET /v1/users/me
API-->>UsersClient: Return User Information
UsersClient-->>Client: Return User Object
Possibly related PRs
Poem
✨ Finishing Touches
Thank you for using CodeRabbit. We offer it for free to the OSS community and would appreciate your support in helping us grow. If you find it useful, would you consider giving us a shout-out on your favorite social media? 🪧 TipsChatThere are 3 ways to chat with CodeRabbit:
Note: Be mindful of the bot's finite context window. It's strongly recommended to break down tasks such as reading entire modules into smaller chunks. For a focused discussion, use review comments to chat about specific files and their changes, instead of using the PR comments. CodeRabbit Commands (Invoked using PR comments)
Other keywords and placeholders
CodeRabbit Configuration File (
|
Codecov ReportAll modified and coverable lines are covered by tests ✅
@@ Coverage Diff @@
## main #177 +/- ##
==========================================
+ Coverage 89.89% 90.01% +0.12%
==========================================
Files 63 65 +2
Lines 5768 5840 +72
==========================================
+ Hits 5185 5257 +72
Misses 583 583
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
🧹 Nitpick comments (6)
cozepy/users/__init__.py (3)
9-13: Consider adding field validations.The
Usermodel could benefit from field validations to ensure data integrity:
user_id: Format validation (e.g., UUID, specific pattern)avatar_url: URL format validationclass User(CozeModel): - user_id: str - user_name: str - nick_name: str - avatar_url: str + user_id: str = Field(pattern=r'^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$') + user_name: str = Field(min_length=1) + nick_name: str = Field(min_length=1) + avatar_url: str = Field(pattern=r'^https?://.+')
22-25: Add docstring to document the method behavior and exceptions.The
memethod should document its behavior, parameters, return type, and possible exceptions.def me(self, **kwargs) -> User: + """Retrieve the current authorized user's information. + + Args: + **kwargs: Additional request parameters. + Supported keys: + - headers (Optional[dict]): Additional request headers. + + Returns: + User: The current user's information. + + Raises: + CozeAPIError: If the API request fails. + ValueError: If the response cannot be parsed. + """ url = f"{self._base_url}/v1/users/me" headers: Optional[dict] = kwargs.get("headers") return self._requester.request("get", url, False, User, headers=headers)
34-37: Add docstring and consider reducing code duplication.
- Add docstring to document the method behavior and exceptions.
- Consider extracting common code between sync and async clients into a base class or mixin.
async def me(self, **kwargs) -> User: + """Retrieve the current authorized user's information asynchronously. + + Args: + **kwargs: Additional request parameters. + Supported keys: + - headers (Optional[dict]): Additional request headers. + + Returns: + User: The current user's information. + + Raises: + CozeAPIError: If the API request fails. + ValueError: If the response cannot be parsed. + """ 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)Consider creating a base class to reduce duplication:
class BaseUsersClient: 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 _get_me_url(self) -> str: return f"{self._base_url}/v1/users/me"tests/test_users.py (3)
9-24: Enhance test coverage with edge cases.Consider adding test cases for:
- Error responses (401, 403, 404, 500)
- Malformed responses
- Empty or null fields
def mock_retrieve_users_me_error(respx_mock, status_code: int) -> httpx.Response: response = httpx.Response( status_code, json={"error": {"code": "unauthorized", "message": "Invalid token"}}, headers={logid_key(): random_hex(10)}, ) respx_mock.get("/v1/users/me").mock(response) return response
27-38: Add more assertions for comprehensive testing.Consider adding assertions for:
- All user fields (user_name, nick_name, avatar_url)
- Response headers
- Raw response data structure
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 + assert user.user_name == mock_user.user_name + assert user.nick_name == mock_user.nick_name + assert user.avatar_url == mock_user.avatar_url + assert user.model_dump() == mock_user.model_dump() + assert user.response.status_code == 200
40-51: Add more assertions and consider reducing test code duplication.
- Add assertions for all user fields and response data.
- Consider using a base test class or fixtures to reduce duplication between sync and async tests.
@pytest.fixture def expected_assertions(): def _assert_user(actual_user: User, mock_user: User): assert actual_user assert actual_user.response.logid == mock_user.response.logid assert actual_user.user_id == mock_user.user_id assert actual_user.user_name == mock_user.user_name assert actual_user.nick_name == mock_user.nick_name assert actual_user.avatar_url == mock_user.avatar_url assert actual_user.model_dump() == mock_user.model_dump() assert actual_user.response.status_code == 200 return _assert_user
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (5)
cozepy/__init__.py(2 hunks)cozepy/coze.py(5 hunks)cozepy/users/__init__.py(1 hunks)examples/users_me.py(1 hunks)tests/test_users.py(1 hunks)
⏰ Context from checks skipped due to timeout of 90000ms (2)
- GitHub Check: test (Python 3.8 on macOS)
- GitHub Check: test (Python 3.8 on Windows)
🔇 Additional comments (8)
examples/users_me.py (2)
15-23: LGTM!The function is well-implemented with good documentation and proper fallback handling.
39-42: LGTM!The function is well-implemented with proper level mapping and fallback handling.
cozepy/__init__.py (1)
94-94: LGTM!The changes follow the module's established patterns for imports and exports.
Also applies to: 293-294
cozepy/coze.py (5)
18-18: LGTM! Import follows consistent pattern.The import is correctly placed within the TYPE_CHECKING block and follows the established import style.
46-46: LGTM! Field declaration follows class pattern.The
_usersfield is correctly declared as Optional[UsersClient] and initialized to None, matching the pattern used for other client fields.
134-140: LGTM! Property implementation follows established pattern.The
usersproperty correctly implements lazy initialization and follows the consistent pattern used throughout the class.
165-165: LGTM! Async field declaration follows class pattern.The
_usersfield is correctly declared as Optional[AsyncUsersClient] and initialized to None, matching the pattern used for other async client fields.
254-260: LGTM! Async property implementation follows established pattern.The async
usersproperty correctly implements lazy initialization and follows the consistent pattern used throughout the class.
| if __name__ == "__main__": | ||
| coze = Coze(auth=TokenAuth(get_coze_api_token()), base_url=get_coze_api_base()) | ||
|
|
||
| user = coze.users.me(**kwargs) | ||
| print(user) |
There was a problem hiding this comment.
🛠️ Refactor suggestion
Add error handling for the API call.
Wrap the API call in a try-except block to handle potential errors gracefully.
if __name__ == "__main__":
- coze = Coze(auth=TokenAuth(get_coze_api_token()), base_url=get_coze_api_base())
+ try:
+ coze = Coze(auth=TokenAuth(get_coze_api_token()), base_url=get_coze_api_base())
- user = coze.users.me(**kwargs)
- print(user)
+ user = coze.users.me(**kwargs)
+ print(user)
+ except Exception as e:
+ logging.error(f"Failed to retrieve user info: {str(e)}")
+ sys.exit(1)Committable suggestion skipped: line range outside the PR's diff.
| 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 |
There was a problem hiding this comment.
🛠️ Refactor suggestion
Improve security and error handling.
- Move the client_id to environment variables or configuration.
- Add error handling for OAuth flow failures.
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()
+ client_id = os.getenv("COZE_CLIENT_ID")
+ if not client_id:
+ raise ValueError("COZE_CLIENT_ID environment variable is required")
- 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
+ try:
+ device_oauth_app = DeviceOAuthApp(client_id=client_id, 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
+ except Exception as e:
+ raise RuntimeError(f"Failed to obtain access token: {str(e)}")📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| 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 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() | |
| client_id = os.getenv("COZE_CLIENT_ID") | |
| if not client_id: | |
| raise ValueError("COZE_CLIENT_ID environment variable is required") | |
| try: | |
| device_oauth_app = DeviceOAuthApp(client_id=client_id, 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 | |
| except Exception as e: | |
| raise RuntimeError(f"Failed to obtain access token: {str(e)}") |
Summary by CodeRabbit
Release Notes
New Features
Documentation
Tests