Merge pull request #388 from yjinjo/master

Refactor code for maintenance

src/spaceone/identity/manager/workspace_group_manager.py

@@ -79,6 +79,20 @@ def list_workspace_groups(self, query: dict) -> Tuple[QuerySet, int]:
     def stat_workspace_group(self, query: dict) -> dict:
         return self.workspace_group_model.stat(**query)
 
+    def get_workspace_group_with_users(
+        self, domain_id: str, workspace_group_id: str
+    ) -> Tuple[WorkspaceGroup, List[str]]:
+        workspace_group_vo = self.get_workspace_group(domain_id, workspace_group_id)
+
+        workspace_group_user_ids = []
+        if workspace_group_vo.users:
+            old_users, new_users = self.get_unique_user_ids(
+                domain_id, workspace_group_id, workspace_group_vo.users
+            )
+            workspace_group_user_ids = old_users + new_users
+
+        return workspace_group_vo, workspace_group_user_ids
+
     @staticmethod
     def _check_existence_of_workspace_or_user(
         workspace_vos: QuerySet, workspace_group_vo: WorkspaceGroup

src/spaceone/identity/service/role_binding_service.py

@@ -119,20 +119,7 @@ def create_role_binding(self, params: dict):
         rb_vo = self.role_binding_manager.create_role_binding(params)
 
         if workspace_vo:
-            user_rb_ids = self.role_binding_manager.stat_role_bindings(
-                query={
-                    "distinct": "user_id",
-                    "filter": [
-                        {"k": "workspace_id", "v": workspace_id, "o": "eq"},
-                        {"k": "domain_id", "v": domain_id, "o": "eq"},
-                    ],
-                }
-            ).get("results", [])
-            user_rb_total_count = len(user_rb_ids)
-
-            self.workspace_mgr.update_workspace_by_vo(
-                {"user_count": user_rb_total_count}, workspace_vo
-            )
+            self.update_workspace_user_count(domain_id, workspace_id)
 
         return rb_vo
 
@@ -462,25 +449,35 @@ def _get_latest_role_type(before: str, after: str) -> str:
 
             return after
 
+    def update_workspace_user_count(self, domain_id: str, workspace_id: str) -> None:
+        workspace_vo = self.workspace_mgr.get_workspace(
+            domain_id=domain_id, workspace_id=workspace_id
+        )
+
+        if workspace_vo:
+            user_rb_total_count = self._get_workspace_user_count(
+                domain_id, workspace_id
+            )
+            self.workspace_mgr.update_workspace_by_vo(
+                {"user_count": user_rb_total_count}, workspace_vo
+            )
+
     def delete_role_binding_by_vo(
         self, rb_vo: RoleBinding, domain_id: str, workspace_id: str = None
     ):
         self.role_binding_manager.delete_role_binding_by_vo(rb_vo)
 
         if workspace_id:
-            workspace_vo = self.workspace_mgr.get_workspace(workspace_id, domain_id)
-
-            user_rb_ids = self.role_binding_manager.stat_role_bindings(
-                query={
-                    "distinct": "user_id",
-                    "filter": [
-                        {"k": "workspace_id", "v": workspace_id, "o": "eq"},
-                        {"k": "domain_id", "v": domain_id, "o": "eq"},
-                    ],
-                }
-            ).get("results", [])
-            user_rb_total_count = len(user_rb_ids)
-
-            self.workspace_mgr.update_workspace_by_vo(
-                {"user_count": user_rb_total_count}, workspace_vo
-            )
+            self.update_workspace_user_count(domain_id, workspace_id)
+
+    def _get_workspace_user_count(self, domain_id: str, workspace_id: str) -> int:
+        user_rb_ids = self.role_binding_manager.stat_role_bindings(
+            query={
+                "distinct": "user_id",
+                "filter": [
+                    {"k": "workspace_id", "v": workspace_id, "o": "eq"},
+                    {"k": "domain_id", "v": domain_id, "o": "eq"},
+                ],
+            }
+        ).get("results", [])
+        return len(user_rb_ids)

src/spaceone/identity/service/workspace_group_service.py

@@ -1,6 +1,7 @@
 import logging
 from datetime import datetime
-from typing import Dict, List, Union
+from mongoengine import QuerySet
+from typing import Dict, List, Union, Any
 
 from spaceone.core.error import ERROR_INVALID_PARAMETER, ERROR_NOT_FOUND
 from spaceone.core.service import (
@@ -237,25 +238,15 @@ def get(
         Returns:
             WorkspaceGroupResponse:
         """
-        workspace_group_id = params.workspace_group_id
-        domain_id = params.domain_id
-
-        workspace_group_vo = self.workspace_group_mgr.get_workspace_group(
-            domain_id, workspace_group_id
-        )
-
-        workspace_group_user_ids = []
-        if workspace_group_vo.users:
-            old_users, new_users = self.workspace_group_mgr.get_unique_user_ids(
-                domain_id, workspace_group_id, workspace_group_vo.users
+        workspace_group_vo, workspace_group_user_ids = (
+            self.workspace_group_mgr.get_workspace_group_with_users(
+                params.domain_id, params.workspace_group_id
             )
-
-            workspace_group_user_ids: List[str] = old_users + new_users
-
-        workspace_group_dict = self.add_user_name_and_state_to_users(
-            workspace_group_user_ids, workspace_group_vo, domain_id
         )
-        return WorkspaceGroupResponse(**workspace_group_dict)
+        workspace_group_info = self.add_user_name_and_state_to_users(
+            workspace_group_user_ids, workspace_group_vo, params.domain_id
+        )
+        return WorkspaceGroupResponse(**workspace_group_info)
 
     @transaction(permission="identity:WorkspaceGroup.read", role_types=["DOMAIN_ADMIN"])
     @append_query_filter(["workspace_group_id", "name", "domain_id"])
@@ -283,27 +274,9 @@ def list(
         workspace_group_vos, total_count = (
             self.workspace_group_mgr.list_workspace_groups(query)
         )
-
-        workspace_groups_info = []
-        for workspace_group_vo in workspace_group_vos:
-            workspace_group_users = workspace_group_vo.users or []
-            old_users = list(
-                set(
-                    [user_info["user_id"] for user_info in workspace_group_users]
-                    if workspace_group_users
-                    else []
-                )
-            )
-            new_users = list(
-                set([user_info["user_id"] for user_info in workspace_group_users])
-            )
-
-            workspace_group_user_ids: List[str] = old_users + new_users
-
-            workspace_group_dict = self.add_user_name_and_state_to_users(
-                workspace_group_user_ids, workspace_group_vo, params.domain_id
-            )
-            workspace_groups_info.append(workspace_group_dict)
+        workspace_groups_info = self.get_workspace_groups_info(
+            workspace_group_vos, params.domain_id
+        )
 
         return WorkspaceGroupsResponse(
             results=workspace_groups_info, total_count=total_count
@@ -599,18 +572,18 @@ def add_user(user_info, workspace_group_workspace_id=None):
     def add_user_name_and_state_to_users(
         self,
         workspace_group_user_ids: List[str],
-        workspace_group_info: Union[WorkspaceGroup, Dict[str, str]],
+        workspace_group_info: Union[WorkspaceGroup, Dict[str, Any]],
         domain_id: str,
-    ) -> Dict[str, str]:
+    ) -> Dict[str, Any]:
         """Add user's name and state to users in workspace group.
         Since the user's name and state are not in user of workspace group in database,
         we need to add user's name and state to users in the Application layer.
         Args:
             workspace_group_user_ids: 'List[str]'
-            workspace_group_info: 'Union[WorkspaceGroup, Dict[str, str]]'
+            workspace_group_info: 'Union[WorkspaceGroup, Dict[str, Any]]'
             domain_id: 'str'
         Returns:
-            workspace_group_info: 'Dict[str, str]'
+            workspace_group_info: 'Dict[str, Any]'
         """
 
         def update_user_info(
@@ -653,10 +626,15 @@ def update_user_info(
             updated_users = [
                 update_user_info(user, user_info_map) for user in workspace_group_users
             ]
-            workspace_group_info.users = updated_users
-            return workspace_group_info.to_dict()
+            if isinstance(workspace_group_info, dict):
+                workspace_group_info["users"] = updated_users
+            else:
+                workspace_group_info.users = updated_users
 
-        return {}
+        if not isinstance(workspace_group_info, dict):
+            return workspace_group_info.to_dict()
+        else:
+            return workspace_group_info
 
     def get_users_info_list(
         self,
@@ -720,25 +698,30 @@ def remove_users_from_workspace_group(
                     {"user_count": user_rb_total_count}, workspace_vo
                 )
         else:
-            workspace_vo = self.workspace_mgr.get_workspace(workspace_id, domain_id)
-            if workspace_vo:
-                user_rb_ids = self.rb_mgr.stat_role_bindings(
-                    query={
-                        "distinct": "user_id",
-                        "filter": [
-                            {"k": "workspace_id", "v": workspace_id, "o": "eq"},
-                            {"k": "domain_id", "v": domain_id, "o": "eq"},
-                        ],
-                    }
-                ).get("results", [])
-                user_rb_total_count = len(user_rb_ids)
-
-                self.workspace_mgr.update_workspace_by_vo(
-                    {"user_count": user_rb_total_count}, workspace_vo
-                )
+            self.rb_svc.update_workspace_user_count(domain_id, workspace_id)
 
         return updated_users
 
+    def get_workspace_groups_info(
+        self, workspace_group_vos: QuerySet, domain_id
+    ) -> List[Dict[str, Any]]:
+        workspace_groups_info = []
+        for workspace_group_vo in workspace_group_vos:
+            workspace_group_user_ids = self._get_workspace_group_user_ids(
+                workspace_group_vo
+            )
+            workspace_group_dict = self.add_user_name_and_state_to_users(
+                workspace_group_user_ids, workspace_group_vo, domain_id
+            )
+            workspace_groups_info.append(workspace_group_dict)
+        return workspace_groups_info
+
+    @staticmethod
+    def _get_workspace_group_user_ids(workspace_group_vo: WorkspaceGroup) -> List[str]:
+        workspace_group_users = workspace_group_vo.users or []
+        user_ids = set(user_info["user_id"] for user_info in workspace_group_users)
+        return list(user_ids)
+
     @staticmethod
     def check_user_state(old_user_id: str, old_user_state: str) -> None:
         if old_user_state in ["DISABLED", "DELETED"]:

src/spaceone/identity/service/workspace_group_user_service.py

@@ -194,24 +194,15 @@ def get(
         Returns:
             WorkspaceGroupResponse:
         """
-        workspace_group_id = params.workspace_group_id
-        domain_id = params.domain_id
-
-        workspace_group_vo = self.workspace_group_mgr.get_workspace_group(
-            domain_id, workspace_group_id
-        )
-
-        workspace_group_user_ids = []
-        if workspace_group_vo.users:
-            old_users, new_users = self.workspace_group_mgr.get_unique_user_ids(
-                domain_id, workspace_group_id, workspace_group_vo.users
+        workspace_group_vo, workspace_group_user_ids = (
+            self.workspace_group_mgr.get_workspace_group_with_users(
+                params.domain_id, params.workspace_group_id
             )
-
-            workspace_group_user_ids: List[str] = old_users + new_users
+        )
 
         workspace_group_dict = (
             self.workspace_group_svc.add_user_name_and_state_to_users(
-                workspace_group_user_ids, workspace_group_vo, domain_id
+                workspace_group_user_ids, workspace_group_vo, params.domain_id
             )
         )
         return WorkspaceGroupResponse(**workspace_group_dict)
@@ -242,31 +233,9 @@ def list(
         workspace_group_vos, total_count = (
             self.workspace_group_mgr.list_workspace_groups(query)
         )
-
-        workspace_groups_info = []
-        for workspace_group_vo in workspace_group_vos:
-            workspace_group_users = workspace_group_vo.users or []
-            old_users = list(
-                set(
-                    [user_info["user_id"] for user_info in workspace_group_users]
-                    if workspace_group_users
-                    else []
-                )
-            )
-            new_users = list(
-                set([user_info["user_id"] for user_info in workspace_group_users])
-            )
-
-            workspace_group_user_ids: List[str] = old_users + new_users
-
-            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)
+        workspace_groups_info = self.workspace_group_svc.get_workspace_groups_info(
+            workspace_group_vos, params.domain_id
+        )
 
         return WorkspaceGroupsResponse(
             results=workspace_groups_info, total_count=total_count