From 2d86bfc45776a18ea89535d3576dd9d0c40f14e3 Mon Sep 17 00:00:00 2001 From: Attack825 <2707138687@qq.com> Date: Fri, 1 Aug 2025 15:27:27 +0800 Subject: [PATCH] feat: imporve user sdk --- src/casdoor/user.py | 139 ++++++++++++++++++++++++++++++++++++++--- src/tests/test_user.py | 67 +++++++++++++++----- 2 files changed, 183 insertions(+), 23 deletions(-) diff --git a/src/casdoor/user.py b/src/casdoor/user.py index 4e61449..4ba284a 100644 --- a/src/casdoor/user.py +++ b/src/casdoor/user.py @@ -13,7 +13,7 @@ # limitations under the License. import json -from typing import Dict, List +from typing import Dict, List, Optional import requests @@ -54,16 +54,18 @@ def __init__(self): self.invitationCode = "" @classmethod - def new(cls, owner, name, created_time, display_name): + def new(cls, owner, name, created_time, display_name, email="", phone=""): self = cls() self.name = name self.owner = owner self.createdTime = created_time self.displayName = display_name + self.email = email + self.phone = phone return self @classmethod - def from_dict(cls, data: dict): + def from_dict(cls, data: dict) -> Optional["User"]: if data is None: return None @@ -79,9 +81,52 @@ def __str__(self): def to_dict(self) -> dict: return self.__dict__ + def get_id(self) -> str: + return f"{self.owner}/{self.name}" + class _UserSDK: - def get_users(self) -> List[Dict]: + def get_global_users(self) -> List[User]: + """ """ + url = self.endpoint + "/api/get-global-users" + params = { + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + r = requests.get(url, params) + response = r.json() + if response["status"] != "ok": + raise Exception(response["msg"]) + users = [] + for user in response["data"]: + users.append(User.from_dict(user)) + return users + + def get_sorted_users(self, sorter: str, limit: str) -> List[User]: + """ + Get the sorted users from Casdoor. + + :param sroter: the DB column name to sort by, e.g., created_time + :param limiter: the count of users to return, e.g., 25 + """ + url = self.endpoint + "/api/get-sorted-users" + params = { + "owner": self.org_name, + "sorter": sorter, + "limit": limit, + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + r = requests.get(url, params) + response = r.json() + if response["status"] != "ok": + raise Exception(response["msg"]) + users = [] + for user in response["data"]: + users.append(User.from_dict(user)) + return users + + def get_users(self) -> List[User]: """ Get the users from Casdoor. @@ -102,16 +147,73 @@ def get_users(self) -> List[Dict]: users.append(User.from_dict(user)) return users - def get_user(self, user_id: str) -> User: + def get_user(self, name: str) -> User: """ - Get the user from Casdoor providing the user_id. + Get the user from Casdoor providing the name. - :param user_id: the id of the user + :param name: the name of the user :return: a dict that contains user's info """ url = self.endpoint + "/api/get-user" params = { - "id": f"{self.org_name}/{user_id}", + "id": f"{self.org_name}/{name}", + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + r = requests.get(url, params) + response = r.json() + if response["status"] != "ok": + raise Exception(response["msg"]) + return User.from_dict(response["data"]) + + def get_user_by_email(self, email: str) -> User: + """ + Get the user from Casdoor providing the email. + + :param email: the email of the user + :return: a User object that contains user's info + """ + url = self.endpoint + "/api/get-user" + params = { + "email": email, + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + r = requests.get(url, params) + response = r.json() + if response["status"] != "ok": + raise Exception(response["msg"]) + return User.from_dict(response["data"]) + + def get_user_by_phone(self, phone: str) -> User: + """ + Get the user from Casdoor providing the phone number. + + :param phone: the phone number of the user + :return: a User object that contains user's info + """ + url = self.endpoint + "/api/get-user" + params = { + "phone": phone, + "clientId": self.client_id, + "clientSecret": self.client_secret, + } + r = requests.get(url, params) + response = r.json() + if response["status"] != "ok": + raise Exception(response["msg"]) + return User.from_dict(response["data"]) + + def get_user_by_user_id(self, user_id: str) -> User: + """ + Get the user from Casdoor providing the user ID. + + :param user_id: the user ID of the user + :return: a User object that contains user's info + """ + url = self.endpoint + "/api/get-user" + params = { + "userId": user_id, "clientId": self.client_id, "clientSecret": self.client_secret, } @@ -146,10 +248,25 @@ def get_user_count(self, is_online: bool = None) -> int: return count def modify_user(self, method: str, user: User) -> Dict: + """ + modifyUser is an encapsulation of user CUD(Create, Update, Delete) operations. + possible actions are `add-user`, `update-user`, `delete-user`, + """ + id = user.get_id() + return self.modify_user_by_id(method, id, user) + + def modify_user_by_id(self, method: str, id: str, user: User) -> Dict: + """ + Modify the user from Casdoor providing the ID. + + :param id: the id ( owner/name ) of the user + :param user: a User object that contains user's info + """ + url = self.endpoint + f"/api/{method}" user.owner = self.org_name params = { - "id": f"{user.owner}/{user.name}", + "id": id, "clientId": self.client_id, "clientSecret": self.client_secret, } @@ -168,6 +285,10 @@ def update_user(self, user: User) -> Dict: response = self.modify_user("update-user", user) return response + def update_user_by_id(self, id: str, user: User) -> Dict: + response = self.modify_user_by_id("update-user", id, user) + return response + def delete_user(self, user: User) -> Dict: response = self.modify_user("delete-user", user) return response diff --git a/src/tests/test_user.py b/src/tests/test_user.py index 2d599f1..9c739c8 100644 --- a/src/tests/test_user.py +++ b/src/tests/test_user.py @@ -15,29 +15,40 @@ import datetime import unittest +import src.tests.test_util as test_util from src.casdoor import CasdoorSDK from src.casdoor.user import User -from src.tests.test_util import ( - TestApplication, - TestClientId, - TestClientSecret, - TestEndpoint, - TestJwtPublicKey, - TestOrganization, - get_random_name, -) class UserTest(unittest.TestCase): + @staticmethod + def get_sdk(): + sdk = CasdoorSDK( + endpoint=test_util.TestEndpoint, + client_id=test_util.TestClientId, + client_secret=test_util.TestClientSecret, + certificate=test_util.TestJwtPublicKey, + org_name=test_util.TestOrganization, + application_name=test_util.TestApplication, + ) + return sdk + def test_user(self): - name = get_random_name("User") + name = test_util.get_random_name("User") + email = f"{name}@gmail.com" + phone = test_util.get_random_code(11) # Add a new object - user = User.new(owner="admin", name=name, created_time=datetime.datetime.now().isoformat(), display_name=name) - - sdk = CasdoorSDK( - TestEndpoint, TestClientId, TestClientSecret, TestJwtPublicKey, TestOrganization, TestApplication + user = User.new( + owner="admin", + name=name, + created_time=datetime.datetime.now().isoformat(), + display_name=name, + email=email, + phone=phone, ) + + sdk = UserTest.get_sdk() try: sdk.add_user(user=user) except Exception as e: @@ -54,9 +65,16 @@ def test_user(self): # Get the object try: user = sdk.get_user(name) + user_id = user.id + user_by_email = sdk.get_user_by_email(email) + user_by_phone = sdk.get_user_by_phone(phone) + user_by_user_id = sdk.get_user_by_user_id(user_id) except Exception as e: self.fail(f"Failed to get object: {e}") self.assertEqual(user.name, name) + self.assertEqual(user_by_email.name, name) + self.assertEqual(user_by_phone.name, name) + self.assertEqual(user_by_user_id.name, name) # Update the object updated_display_name = "Updated Casdoor Website" @@ -85,3 +103,24 @@ def test_user(self): except Exception as e: self.fail(f"Failed to get object: {e}") self.assertIsNone(deleted_user, "Failed to delete object, it's still retrievable") + + def test_get_global_users(self): + sdk = UserTest.get_sdk() + try: + users = sdk.get_global_users() + except Exception as e: + self.fail(f"Fail to get object:{e}") + + self.assertIsInstance(users, list, "The returned result is not a list") + for user in users: + self.assertIsInstance(user, User, "There are non User type objects in the list") + + def test_get_sort_users(self): + sdk = UserTest.get_sdk() + try: + users = sdk.get_sorted_users("created_time", 25) + except Exception as e: + self.fail(f"Fail to get object:{e}") + self.assertIsInstance(users, list, "The returned result is not a list") + for user in users: + self.assertIsInstance(user, User, "There are non User type objects in the list")