feat: refactor code and modify get_workspace_groups

Signed-off-by: Youngjin Jo <[email protected]>

src/spaceone/identity/manager/workspace_group_manager.py

@@ -125,18 +125,23 @@ def check_user_id_in_users(user_id: str, workspace_group_vo: WorkspaceGroup):
         )
 
     def get_old_users_and_new_users(
-        self, users: List[Dict[str, str]], workspace_group_id: str, domain_id: str
+        self,
+        new_users_info_list: List[Dict[str, str]],
+        workspace_group_id: str,
+        domain_id: str,
     ) -> Tuple[List[str], List[str]]:
         workspace_group_vo = self.get_workspace_group(workspace_group_id, domain_id)
 
         old_users = list(
             set(
-                [user_info["user_id"] for user_info in workspace_group_vo.users]
+                [old_user_info["user_id"] for old_user_info in workspace_group_vo.users]
                 if workspace_group_vo.users
                 else []
             )
         )
-        new_users = list(set([user_info["user_id"] for user_info in users]))
+        new_users = list(
+            set([new_user_info["user_id"] for new_user_info in new_users_info_list])
+        )
 
         return old_users, new_users
 

src/spaceone/identity/service/user_profile_service.py

@@ -50,6 +50,7 @@ def __init__(self, *args, **kwargs):
         self.domain_mgr = DomainManager()
         self.domain_secret_mgr = DomainSecretManager()
         self.workspace_group_mgr = WorkspaceGroupManager()
+        self.workspace_group_svc = WorkspaceGroupService()
 
     @transaction(permission="identity:UserProfile.write", role_types=["USER"])
     @convert_model
@@ -448,17 +449,18 @@ def get_workspace_groups(
         )
         role_bindings_info_map = {rb.workspace_group_id: rb.to_dict() for rb in rb_vos}
 
-        workspace_group_user_ids = [
-            user["user_id"]
-            for workspace_group in workspace_group_vos
-            for user in workspace_group["users"]
-        ]
+        workspace_group_user_ids = []
+        for workspace_group in workspace_group_vos:
+            if workspace_group.users:
+                for user in workspace_group.users:
+                    workspace_group_user_ids.append(user.user_id)
 
-        workspace_group_svc = WorkspaceGroupService()
         workspace_groups_info = []
         for workspace_group_vo in workspace_group_vos:
-            workspace_group_dict = workspace_group_svc.add_user_name_and_state_to_users(
-                workspace_group_user_ids, workspace_group_vo, params.domain_id
+            workspace_group_dict = (
+                self.workspace_group_svc.add_user_name_and_state_to_users(
+                    workspace_group_user_ids, workspace_group_vo, params.domain_id
+                )
             )
             workspace_groups_info.append(workspace_group_dict)
 

src/spaceone/identity/service/workspace_group_service.py

@@ -60,6 +60,7 @@ class WorkspaceGroupService(BaseService):
 
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
+        self.workspace_mgr = WorkspaceManager()
         self.workspace_group_mgr = WorkspaceGroupManager()
         self.user_mgr = UserManager()
         self.role_mgr = RoleManager()
@@ -100,14 +101,14 @@ def create(
     def update(
         self, params: WorkspaceGroupUpdateRequest
     ) -> Union[WorkspaceGroupResponse, dict]:
-        """Update workspace group
+        """Update workspace group name and tags
         Args:
             params (WorkspaceGroupUpdateRequest): {
                 'workspace_group_id': 'str',    # required
                 'name': 'str',
                 'tags': 'dict',
                 'workspace_id': 'str',          # injected from auth
-                'domain_id': 'str',             # injected from auth
+                'domain_id': 'str',             # injected from auth (required)
         Returns:
             WorkspaceGroupResponse:
         """
@@ -159,8 +160,8 @@ def add_users(
                 'workspace_group_id': 'str',      # required
                 'users': [                        # required
                     {
-                        'user_id': 'str',
-                        'role_id': 'str',
+                        'user_id': 'str',         # required
+                        'role_id': 'str',         # required
                     }
                 ],
                 'workspace_id': 'str',            # injected from auth
@@ -173,15 +174,15 @@ def add_users(
         if role_type != "DOMAIN_ADMIN":
             raise ERROR_PERMISSION_DENIED()
 
-        users: List[Dict[str, str]] = params.users
+        new_users_info_list: List[Dict[str, str]] = params.users
         workspace_group_id = params.workspace_group_id
         domain_id = params.domain_id
         workspace_group_vo = self.workspace_group_mgr.get_workspace_group(
             workspace_group_id, domain_id
         )
 
         old_users, new_users = self.workspace_group_mgr.get_old_users_and_new_users(
-            users, workspace_group_id, domain_id
+            new_users_info_list, workspace_group_id, domain_id
         )
 
         self.workspace_group_mgr.check_new_users_already_in_workspace_group(
@@ -191,23 +192,29 @@ def add_users(
 
         self.check_new_users_exist_in_domain(new_users, domain_id)
 
-        role_map = self.get_role_map(users, domain_id)
-        workspace_ids = self.get_workspace_ids(workspace_group_id, domain_id)
+        role_map = self.get_role_map(new_users_info_list, domain_id)
+        workspace_ids_in_workspace_group = self.get_workspace_ids(
+            workspace_group_id, domain_id
+        )
         old_users_in_workspace_group = workspace_group_vo.users or []
         new_users_in_workspace_group = self.add_users_to_workspace_group(
-            users, role_map, workspace_ids, workspace_group_id, domain_id
+            new_users_info_list,
+            role_map,
+            workspace_ids_in_workspace_group,
+            workspace_group_id,
+            domain_id,
         )
         params.users = old_users_in_workspace_group + new_users_in_workspace_group
 
         workspace_group_vo = self.workspace_group_mgr.update_workspace_group_by_vo(
             params.dict(exclude_unset=True), workspace_group_vo
         )
 
-        workspace_group_dict = self.add_user_name_and_state_to_users(
+        workspace_group_info = self.add_user_name_and_state_to_users(
             workspace_group_user_ids, workspace_group_vo, domain_id
         )
 
-        return WorkspaceGroupResponse(**workspace_group_dict)
+        return WorkspaceGroupResponse(**workspace_group_info)
 
     @transaction(
         permission="identity:WorkspaceGroup.write", role_types=["DOMAIN_ADMIN"]
@@ -438,7 +445,9 @@ def stat(self, params: WorkspaceGroupStatQueryRequest) -> dict:
 
         return self.workspace_group_mgr.stat_workspace_group(query)
 
-    def check_new_users_exist_in_domain(self, new_users: List[str], domain_id: str):
+    def check_new_users_exist_in_domain(
+        self, new_users: List[str], domain_id: str
+    ) -> None:
         user_vos = self.user_mgr.filter_users(user_id=new_users, domain_id=domain_id)
         if not user_vos.count() == len(new_users):
             raise ERROR_NOT_FOUND(key="user_id", value=new_users)
@@ -462,10 +471,8 @@ def get_role_map(
 
         return role_map
 
-    @staticmethod
-    def get_workspace_ids(workspace_group_id: str, domain_id: str) -> List[str]:
-        workspace_mgr = WorkspaceManager()
-        workspace_vos = workspace_mgr.filter_workspaces(
+    def get_workspace_ids(self, workspace_group_id: str, domain_id: str) -> List[str]:
+        workspace_vos = self.workspace_mgr.filter_workspaces(
             workspace_group_id=workspace_group_id, domain_id=domain_id
         )
         workspace_ids = [workspace_vo.workspace_id for workspace_vo in workspace_vos]
@@ -544,19 +551,19 @@ def add_user_name_and_state_to_users(
                 "state": user_vo.state,
             }
 
-        workspace_group_dict = workspace_group_vo.to_dict()
+        workspace_group_info = workspace_group_vo.to_dict()
 
-        if workspace_group_dict.get("users", []) is not None:
+        if workspace_group_info.get("users", []) is not None:
             users = []
-            for user in workspace_group_dict["users"]:
+            for user in workspace_group_info["users"]:
                 user_id = user["user_id"]
                 user["user_name"] = user_info_map[user_id]["name"]
                 user["state"] = user_info_map[user_id]["state"]
                 users.append(user)
 
-            workspace_group_dict["users"] = users
+            workspace_group_info["users"] = users
 
-        return workspace_group_dict
+        return workspace_group_info
 
     def remove_users_from_workspace_group(
         self,

src/spaceone/identity/service/workspace_group_user_service.py

@@ -269,7 +269,41 @@ def find(
         Returns:
             WorkspaceGroupUsersSummaryResponse:
         """
-        return self._find(params)
+        workspace_group = self.workspace_group_mgr.get_workspace_group(
+            params.workspace_group_id, params.domain_id
+        )
+        workspace_group_users = workspace_group.users or []
+        workspace_group_user_ids = list(
+            set([user["user_id"] for user in workspace_group_users])
+        )
+
+        query = {
+            "filter": [
+                {"k": "domain_id", "v": params.domain_id, "o": "eq"},
+                {"k": "user_id", "v": workspace_group_user_ids, "o": "not_in"},
+            ],
+            "sort": [{"key": "user_id"}],
+            "page": params.page,
+            "only": ["user_id", "name", "state"],
+        }
+
+        if params.keyword:
+            query["filter_or"] = [
+                {"k": "user_id", "v": params.keyword, "o": "contain"},
+                {"k": "name", "v": params.keyword, "o": "contain"},
+            ]
+
+        if params.state:
+            query["filter"].append({"k": "state", "v": params.state, "o": "eq"})
+
+        user_mgr = UserManager()
+        user_vos, total_count = user_mgr.list_users(query)
+
+        workspace_group_users_info = [user_vo.to_dict() for user_vo in user_vos]
+
+        return WorkspaceGroupUsersSummaryResponse(
+            results=workspace_group_users_info, total_count=total_count
+        )
 
     @transaction(permission="identity:WorkspaceGroupUser:read", role_types=["USER"])
     @convert_model
@@ -383,42 +417,3 @@ def stat(self, params: WorkspaceGroupUserStatQueryRequest) -> dict:
         return self.workspace_group_user_mgr.stat_workspace_group_users(
             query, params.workspace_group_id, params.domain_id
         )
-
-    def _find(
-        self, params: WorkspaceGroupUserFindRequest
-    ) -> Union[WorkspaceGroupUsersSummaryResponse, dict]:
-        workspace_group = self.workspace_group_mgr.get_workspace_group(
-            params.workspace_group_id, params.domain_id
-        )
-        workspace_group_users = workspace_group.users or []
-        workspace_group_user_ids = list(
-            set([user["user_id"] for user in workspace_group_users])
-        )
-
-        query = {
-            "filter": [
-                {"k": "domain_id", "v": params.domain_id, "o": "eq"},
-                {"k": "user_id", "v": workspace_group_user_ids, "o": "not_in"},
-            ],
-            "sort": [{"key": "user_id"}],
-            "page": params.page,
-            "only": ["user_id", "name", "state"],
-        }
-
-        if params.keyword:
-            query["filter_or"] = [
-                {"k": "user_id", "v": params.keyword, "o": "contain"},
-                {"k": "name", "v": params.keyword, "o": "contain"},
-            ]
-
-        if params.state:
-            query["filter"].append({"k": "state", "v": params.state, "o": "eq"})
-
-        user_mgr = UserManager()
-        user_vos, total_count = user_mgr.list_users(query)
-
-        workspace_group_users_info = [user_vo.to_dict() for user_vo in user_vos]
-
-        return WorkspaceGroupUsersSummaryResponse(
-            results=workspace_group_users_info, total_count=total_count
-        )