two version of R2R are hereHEADmaster

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)