diff options
Diffstat (limited to '.venv/lib/python3.12/site-packages/sdk/sync_methods/users.py')
| -rw-r--r-- | .venv/lib/python3.12/site-packages/sdk/sync_methods/users.py | 587 |
1 files changed, 587 insertions, 0 deletions
diff --git a/.venv/lib/python3.12/site-packages/sdk/sync_methods/users.py b/.venv/lib/python3.12/site-packages/sdk/sync_methods/users.py new file mode 100644 index 00000000..3021642f --- /dev/null +++ b/.venv/lib/python3.12/site-packages/sdk/sync_methods/users.py @@ -0,0 +1,587 @@ +from typing import Any, Optional +from uuid import UUID + +from shared.api.models import ( + WrappedAPIKeyResponse, + WrappedAPIKeysResponse, + WrappedBooleanResponse, + WrappedCollectionsResponse, + WrappedGenericMessageResponse, + WrappedLimitsResponse, + WrappedLoginResponse, + WrappedTokenResponse, + WrappedUserResponse, + WrappedUsersResponse, +) + + +class UsersSDK: + def __init__(self, client): + self.client = client + + def create( + self, + email: str, + password: str, + name: Optional[str] = None, + bio: Optional[str] = None, + profile_picture: Optional[str] = None, + ) -> WrappedUserResponse: + """Register a new user. + + Args: + email (str): User's email address + password (str): User's password + name (Optional[str]): The name for the new user + bio (Optional[str]): The bio for the new user + profile_picture (Optional[str]): New user profile picture + + Returns: + UserResponse: New user information + """ + + data: dict = {"email": email, "password": password} + + if name is not None: + data["name"] = name + if bio is not None: + data["bio"] = bio + if profile_picture is not None: + data["profile_picture"] = profile_picture + + response_dict = self.client._make_request( + "POST", + "users", + json=data, + version="v3", + ) + + return WrappedUserResponse(**response_dict) + + def send_verification_email( + self, email: str + ) -> WrappedGenericMessageResponse: + """Request that a verification email to a user.""" + response_dict = self.client._make_request( + "POST", + "users/send-verification-email", + json=email, + version="v3", + ) + + return WrappedGenericMessageResponse(**response_dict) + + def delete(self, id: str | UUID, password: str) -> WrappedBooleanResponse: + """Delete a specific user. Users can only delete their own account + unless they are superusers. + + Args: + id (str | UUID): User ID to delete + password (str): User's password + + Returns: + dict: Deletion result + """ + data: dict[str, Any] = {"password": password} + response_dict = self.client._make_request( + "DELETE", + f"users/{str(id)}", + json=data, + version="v3", + ) + self.client.access_token = None + self.client._refresh_token = None + + return WrappedBooleanResponse(**response_dict) + + def verify_email( + self, email: str, verification_code: str + ) -> WrappedGenericMessageResponse: + """Verify a user's email address. + + Args: + email (str): User's email address + verification_code (str): Verification code sent to the user's email + + Returns: + dict: Verification result + """ + data: dict[str, Any] = { + "email": email, + "verification_code": verification_code, + } + response_dict = self.client._make_request( + "POST", + "users/verify-email", + json=data, + version="v3", + ) + + return WrappedGenericMessageResponse(**response_dict) + + def login(self, email: str, password: str) -> WrappedLoginResponse: + """Log in a user. + + Args: + email (str): User's email address + password (str): User's password + + Returns: + WrappedLoginResponse + """ + if self.client.api_key: + raise ValueError( + "Cannot log in after setting an API key, please unset your R2R_API_KEY variable or call client.set_api_key(None)" + ) + data: dict[str, Any] = {"username": email, "password": password} + response_dict = self.client._make_request( + "POST", + "users/login", + data=data, + version="v3", + ) + + login_response = WrappedLoginResponse(**response_dict) + self.client.access_token = login_response.results.access_token.token + self.client._refresh_token = login_response.results.refresh_token.token + + user = self.client._make_request( + "GET", + "users/me", + version="v3", + ) + + user_response = WrappedUserResponse(**user) + self.client._user_id = user_response.results.id + + return login_response + + def logout(self) -> WrappedGenericMessageResponse | None: + """Log out the current user.""" + if self.client.access_token: + response_dict = self.client._make_request( + "POST", + "users/logout", + version="v3", + ) + self.client.access_token = None + self.client._refresh_token = None + + return WrappedGenericMessageResponse(**response_dict) + + self.client.access_token = None + self.client._refresh_token = None + return None + + def refresh_token(self) -> WrappedTokenResponse: + """Refresh the access token using the refresh token.""" + if self.client._refresh_token: + response_dict = self.client._make_request( + "POST", + "users/refresh-token", + json=self.client._refresh_token, + version="v3", + ) + + self.client.access_token = response_dict["results"]["access_token"][ + "token" + ] + self.client._refresh_token = response_dict["results"]["refresh_token"][ + "token" + ] + + return WrappedTokenResponse(**response_dict) + + def change_password( + self, current_password: str, new_password: str + ) -> WrappedGenericMessageResponse: + """Change the user's password. + + Args: + current_password (str): User's current password + new_password (str): User's new password + + Returns: + dict: Change password result + """ + data: dict[str, Any] = { + "current_password": current_password, + "new_password": new_password, + } + response_dict = self.client._make_request( + "POST", + "users/change-password", + json=data, + version="v3", + ) + + return WrappedGenericMessageResponse(**response_dict) + + def request_password_reset( + self, email: str + ) -> WrappedGenericMessageResponse: + """Request a password reset. + + Args: + email (str): User's email address + + Returns: + dict: Password reset request result + """ + response_dict = self.client._make_request( + "POST", + "users/request-password-reset", + json=email, + version="v3", + ) + + return WrappedGenericMessageResponse(**response_dict) + + def reset_password( + self, reset_token: str, new_password: str + ) -> WrappedGenericMessageResponse: + """Reset password using a reset token. + + Args: + reset_token (str): Password reset token + new_password (str): New password + + Returns: + dict: Password reset result + """ + data: dict[str, Any] = { + "reset_token": reset_token, + "new_password": new_password, + } + response_dict = self.client._make_request( + "POST", + "users/reset-password", + json=data, + version="v3", + ) + + return WrappedGenericMessageResponse(**response_dict) + + def list( + self, + ids: Optional[list[str | UUID]] = None, + offset: Optional[int] = 0, + limit: Optional[int] = 100, + ) -> WrappedUsersResponse: + """List users with pagination and filtering options. + + Args: + offset (int, optional): Specifies the number of objects to skip. Defaults to 0. + limit (int, optional): Specifies a limit on the number of objects to return, ranging between 1 and 100. Defaults to 100. + + Returns: + dict: List of users and pagination information + """ + params = { + "offset": offset, + "limit": limit, + } + if ids: + params["ids"] = [str(user_id) for user_id in ids] # type: ignore + + response_dict = self.client._make_request( + "GET", + "users", + params=params, + version="v3", + ) + + return WrappedUsersResponse(**response_dict) + + def retrieve( + self, + id: str | UUID, + ) -> WrappedUserResponse: + """Get a specific user. + + Args: + id (str | UUID): User ID to retrieve + + Returns: + dict: Detailed user information + """ + response_dict = self.client._make_request( + "GET", + f"users/{str(id)}", + version="v3", + ) + + return WrappedUserResponse(**response_dict) + + def me( + self, + ) -> WrappedUserResponse: + """Get detailed information about the currently authenticated user. + + Returns: + dict: Detailed user information + """ + response_dict = self.client._make_request( + "GET", + "users/me", + version="v3", + ) + + return WrappedUserResponse(**response_dict) + + def update( + self, + id: str | UUID, + email: Optional[str] = None, + is_superuser: Optional[bool] = None, + name: Optional[str] = None, + bio: Optional[str] = None, + profile_picture: Optional[str] = None, + limits_overrides: dict | None = None, + metadata: dict[str, str | None] | None = None, + ) -> WrappedUserResponse: + """Update user information. + + Args: + id (str | UUID): User ID to update + username (Optional[str]): New username + is_superuser (Optional[bool]): Update superuser status + name (Optional[str]): New name + bio (Optional[str]): New bio + profile_picture (Optional[str]): New profile picture + + Returns: + dict: Updated user information + """ + data: dict = {} + if email is not None: + data["email"] = email + if is_superuser is not None: + data["is_superuser"] = is_superuser + if name is not None: + data["name"] = name + if bio is not None: + data["bio"] = bio + if profile_picture is not None: + data["profile_picture"] = profile_picture + if limits_overrides is not None: + data["limits_overrides"] = limits_overrides + if metadata is not None: + data["metadata"] = metadata + + response_dict = self.client._make_request( + "POST", + f"users/{str(id)}", + json=data, + version="v3", + ) + + return WrappedUserResponse(**response_dict) + + def list_collections( + self, + id: str | UUID, + offset: Optional[int] = 0, + limit: Optional[int] = 100, + ) -> WrappedCollectionsResponse: + """Get all collections associated with a specific user. + + Args: + id (str | UUID): User ID to get collections for + offset (int, optional): Specifies the number of objects to skip. Defaults to 0. + limit (int, optional): Specifies a limit on the number of objects to return, ranging between 1 and 100. Defaults to 100. + + Returns: + dict: List of collections and pagination information + """ + params = { + "offset": offset, + "limit": limit, + } + + response_dict = self.client._make_request( + "GET", + f"users/{str(id)}/collections", + params=params, + version="v3", + ) + + return WrappedCollectionsResponse(**response_dict) + + def add_to_collection( + self, + id: str | UUID, + collection_id: str | UUID, + ) -> WrappedBooleanResponse: + """Add a user to a collection. + + Args: + id (str | UUID): User ID to add + collection_id (str | UUID): Collection ID to add user to + """ + response_dict = self.client._make_request( + "POST", + f"users/{str(id)}/collections/{str(collection_id)}", + version="v3", + ) + + return WrappedBooleanResponse(**response_dict) + + def remove_from_collection( + self, + id: str | UUID, + collection_id: str | UUID, + ) -> WrappedBooleanResponse: + """Remove a user from a collection. + + Args: + id (str | UUID): User ID to remove + collection_id (str | UUID): Collection ID to remove user from + + Returns: + bool: True if successful + """ + response_dict = self.client._make_request( + "DELETE", + f"users/{str(id)}/collections/{str(collection_id)}", + version="v3", + ) + + return WrappedBooleanResponse(**response_dict) + + def create_api_key( + self, + id: str | UUID, + name: Optional[str] = None, + description: Optional[str] = None, + ) -> WrappedAPIKeyResponse: + """Create a new API key for the specified user. + + Args: + id (str | UUID): User ID to create API key for + name (Optional[str]): Name of the API key + description (Optional[str]): Description of the API key + + Returns: + dict: { "message": "API key created successfully", "api_key": "key_id.raw_api_key" } + """ + data: dict[str, Any] = {} + if name: + data["name"] = name + if description: + data["description"] = description + + response_dict = self.client._make_request( + "POST", + f"users/{str(id)}/api-keys", + json=data, + version="v3", + ) + + return WrappedAPIKeyResponse(**response_dict) + + def list_api_keys( + self, + id: str | UUID, + ) -> WrappedAPIKeysResponse: + """List all API keys for the specified user. + + Args: + id (str | UUID): User ID to list API keys for + + Returns: + WrappedAPIKeysResponse + """ + resp_dict = self.client._make_request( + "GET", + f"users/{str(id)}/api-keys", + version="v3", + ) + + return WrappedAPIKeysResponse(**resp_dict) + + def delete_api_key( + self, + id: str | UUID, + key_id: str | UUID, + ) -> WrappedBooleanResponse: + """Delete a specific API key for the specified user. + + Args: + id (str | UUID): User ID + key_id (str | UUID): API key ID to delete + + Returns: + dict: { "message": "API key deleted successfully" } + """ + response_dict = self.client._make_request( + "DELETE", + f"users/{str(id)}/api-keys/{str(key_id)}", + version="v3", + ) + + return WrappedBooleanResponse(**response_dict) + + def get_limits(self) -> WrappedLimitsResponse: + response_dict = self.client._make_request( + "GET", + f"users/{str(self.client._user_id)}/limits", + version="v3", + ) + + return WrappedLimitsResponse(**response_dict) + + def oauth_google_authorize(self) -> WrappedGenericMessageResponse: + """Get Google OAuth 2.0 authorization URL from the server. + + Returns: + WrappedGenericMessageResponse + """ + response_dict = self.client._make_request( + "GET", + "users/oauth/google/authorize", + version="v3", + ) + + return WrappedGenericMessageResponse(**response_dict) + + def oauth_github_authorize(self) -> WrappedGenericMessageResponse: + """Get GitHub OAuth 2.0 authorization URL from the server. + + Returns: {"redirect_url": "..."} + """ + response_dict = self.client._make_request( + "GET", + "users/oauth/github/authorize", + version="v3", + ) + + return WrappedGenericMessageResponse(**response_dict) + + def oauth_google_callback( + self, code: str, state: str + ) -> WrappedLoginResponse: + """Exchange `code` and `state` with the Google OAuth 2.0 callback + route.""" + response_dict = self.client._make_request( + "GET", + "users/oauth/google/callback", + params={"code": code, "state": state}, + version="v3", + ) + + return WrappedLoginResponse(**response_dict) + + def oauth_github_callback( + self, code: str, state: str + ) -> WrappedLoginResponse: + """Exchange `code` and `state` with the GitHub OAuth 2.0 callback + route.""" + response_dict = self.client._make_request( + "GET", + "users/oauth/github/callback", + params={"code": code, "state": state}, + version="v3", + ) + + return WrappedLoginResponse(**response_dict) |