feat: add workspace_count field to workspace_group and refactor the code

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

src/spaceone/identity/manager/workspace_group_manager.py

@@ -27,6 +27,8 @@ def _rollback(vo: WorkspaceGroup):
             )
             vo.delete()
 
+        params["workspace_count"] = 0
+
         workspace_group_vo = self.workspace_group_model.create(params)
         self.transaction.add_rollback(_rollback, workspace_group_vo)
 

src/spaceone/identity/model/workspace_group/database.py

@@ -3,6 +3,7 @@
     DictField,
     EmbeddedDocument,
     EmbeddedDocumentField,
+    IntField,
     ListField,
     StringField,
 )
@@ -20,6 +21,7 @@ class WorkspaceGroupUser(EmbeddedDocument):
 class WorkspaceGroup(MongoModel):
     workspace_group_id = StringField(max_length=40, generate_id="wg", unique=True)
     name = StringField(max_length=255, unique_with="domain_id")
+    workspace_count = IntField(default=None)
     users = ListField(
         EmbeddedDocumentField(WorkspaceGroupUser), default=None, null=True
     )
@@ -33,6 +35,7 @@ class WorkspaceGroup(MongoModel):
     meta = {
         "updatable_fields": [
             "name",
+            "workspace_count",
             "users",
             "tags",
             "updated_by",

src/spaceone/identity/model/workspace_group/response.py

@@ -18,6 +18,7 @@ class WorkspaceGroupUser(BaseModel):
 class WorkspaceGroupResponse(BaseModel):
     workspace_group_id: Union[str, None] = None
     name: Union[str, None] = None
+    workspace_count: Union[int, None] = None
     users: Union[List[WorkspaceGroupUser], None] = None
     tags: Union[dict, None] = None
     created_by: Union[str, None] = None

src/spaceone/identity/service/workspace_group_service.py

@@ -60,11 +60,9 @@ 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()
-        self.rb_svc = RoleBindingService()
         self.rb_mgr = RoleBindingManager()
 
     @transaction(
@@ -195,14 +193,14 @@ def add_users(
         self.check_new_users_exist_in_domain(new_users, 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_workspace_ids = 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(
             new_users_info_list,
             role_map,
-            workspace_ids_in_workspace_group,
+            workspace_group_workspace_ids,
             workspace_group_id,
             domain_id,
         )
@@ -356,13 +354,15 @@ def get(
             workspace_group_id, domain_id
         )
 
-        old_users, new_users = (
-            self.workspace_group_mgr.get_unique_old_users_and_new_users(
-                workspace_group_vo.users, workspace_group_id, domain_id
+        workspace_group_user_ids = []
+        if workspace_group_vo.users:
+            old_users, new_users = (
+                self.workspace_group_mgr.get_unique_old_users_and_new_users(
+                    workspace_group_vo.users, workspace_group_id, domain_id
+                )
             )
-        )
 
-        workspace_group_user_ids: List[str] = old_users + new_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, domain_id
@@ -477,53 +477,56 @@ def get_role_map(
 
         return role_map
 
-    def get_workspace_ids(self, workspace_group_id: str, domain_id: str) -> List[str]:
-        workspace_vos = self.workspace_mgr.filter_workspaces(
+    @staticmethod
+    def get_workspace_ids(workspace_group_id: str, domain_id: str) -> List[str]:
+        workspace_mgr = WorkspaceManager()
+        workspace_vos = 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]
 
         return workspace_ids
 
+    @staticmethod
     def add_users_to_workspace_group(
-        self,
-        users: List[Dict[str, str]],
+        new_users_info_list: List[Dict[str, str]],
         role_map: Dict[str, str],
-        workspace_ids: List[str],
+        workspace_group_workspace_ids: List[str],
         workspace_group_id: str,
         domain_id: str,
-    ):
+    ) -> List[Dict[str, str]]:
         new_users_in_workspace_group = []
-        if workspace_ids:
-            for workspace_id in workspace_ids:
-                for user_info in users:
-                    role_type = role_map[user_info["role_id"]]
+        rb_svc = RoleBindingService()
+        if workspace_group_workspace_ids:
+            for workspace_id in workspace_group_workspace_ids:
+                for new_user_info in new_users_info_list:
+                    role_type = role_map[new_user_info["role_id"]]
 
                     role_binding_params = {
-                        "user_id": user_info["user_id"],
-                        "role_id": user_info["role_id"],
+                        "user_id": new_user_info["user_id"],
+                        "role_id": new_user_info["role_id"],
                         "role_type": role_type,
                         "resource_group": "WORKSPACE",
                         "domain_id": domain_id,
                         "workspace_group_id": workspace_group_id,
                         "workspace_id": workspace_id,
                     }
-                    new_rb_vo = self.rb_svc.create_role_binding(role_binding_params)
+                    new_user_rb_vo = rb_svc.create_role_binding(role_binding_params)
                     new_users_in_workspace_group.append(
                         {
-                            "user_id": new_rb_vo.user_id,
-                            "role_id": new_rb_vo.role_id,
-                            "role_type": new_rb_vo.role_type,
+                            "user_id": new_user_rb_vo.user_id,
+                            "role_id": new_user_rb_vo.role_id,
+                            "role_type": new_user_rb_vo.role_type,
                         }
                     )
         else:
-            for user_info in users:
-                role_type = role_map[user_info["role_id"]]
+            for new_user_info in new_users_info_list:
+                role_type = role_map[new_user_info["role_id"]]
 
                 new_users_in_workspace_group.append(
                     {
-                        "user_id": user_info["user_id"],
-                        "role_id": user_info["role_id"],
+                        "user_id": new_user_info["user_id"],
+                        "role_id": new_user_info["role_id"],
                         "role_type": role_type,
                     }
                 )
@@ -559,7 +562,7 @@ def add_user_name_and_state_to_users(
 
         workspace_group_info = workspace_group_vo.to_dict()
 
-        if workspace_group_info.get("users", []) is not None:
+        if workspace_group_info.get("users", []):
             users = []
             for user in workspace_group_info["users"]:
                 user_id = user["user_id"]

src/spaceone/identity/service/workspace_group_user_service.py

@@ -84,16 +84,18 @@ def add(
            WorkspaceGroupResponse:
         """
         workspace_group_id = params.workspace_group_id
-        users: List[Dict[str, str]] = params.users
+        new_users_info_list: List[Dict[str, str]] = params.users
         user_id = params.user_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
+        old_users, new_users = (
+            self.workspace_group_mgr.get_unique_old_users_and_new_users(
+                new_users_info_list, workspace_group_id, domain_id
+            )
         )
         self.workspace_group_mgr.check_new_users_already_in_workspace_group(
             old_users, new_users
@@ -103,27 +105,27 @@ def add(
 
         self.workspace_group_svc.check_new_users_exist_in_domain(new_users, domain_id)
 
-        old_users_in_workspace_group = workspace_group_vo.users or []
-        if old_users_in_workspace_group:
+        workspace_group_old_users_info = workspace_group_vo.users or []
+        if workspace_group_old_users_info:
             self.workspace_group_user_mgr.check_user_role_type(
-                old_users_in_workspace_group, user_id, command="add"
+                workspace_group_old_users_info, user_id, command="add"
             )
 
-        role_map = self.workspace_group_svc.get_role_map(users, domain_id)
+        role_map = self.workspace_group_svc.get_role_map(new_users_info_list, domain_id)
 
-        workspace_ids = self.workspace_group_svc.get_workspace_ids(
+        workspace_group_workspace_ids = self.workspace_group_svc.get_workspace_ids(
             workspace_group_id, domain_id
         )
-        new_users_in_workspace_group = (
+        workspace_group_new_users_info = (
             self.workspace_group_svc.add_users_to_workspace_group(
-                users,
+                new_users_info_list,
                 role_map,
-                workspace_ids,
+                workspace_group_workspace_ids,
                 workspace_group_id,
                 domain_id,
             )
         )
-        params.users = old_users_in_workspace_group + new_users_in_workspace_group
+        params.users = workspace_group_old_users_info + workspace_group_new_users_info
 
         workspace_group_vo = self.workspace_group_mgr.update_workspace_group_by_vo(
             params.dict(exclude_unset=True), workspace_group_vo
@@ -163,8 +165,10 @@ def remove(
         user_id = params.user_id
         domain_id = params.domain_id
 
-        old_user_ids, user_ids = self.workspace_group_mgr.get_old_users_and_new_users(
-            users, workspace_group_id, domain_id
+        old_user_ids, user_ids = (
+            self.workspace_group_mgr.get_unique_old_users_and_new_users(
+                users, workspace_group_id, domain_id
+            )
         )
         self.workspace_group_mgr.check_user_ids_exist_in_workspace_group(
             old_user_ids, user_ids
@@ -327,11 +331,15 @@ def get(
             workspace_group_id, domain_id
         )
 
-        old_users, new_users = self.workspace_group_mgr.get_old_users_and_new_users(
-            workspace_group_vo.users, workspace_group_id, domain_id
-        )
+        workspace_group_user_ids = []
+        if workspace_group_vo.users:
+            old_users, new_users = (
+                self.workspace_group_mgr.get_unique_old_users_and_new_users(
+                    workspace_group_vo.users, workspace_group_id, domain_id
+                )
+            )
 
-        workspace_group_user_ids: List[str] = old_users + new_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(

src/spaceone/identity/service/workspace_service.py

@@ -11,9 +11,12 @@
 from spaceone.identity.manager.project_manager import ProjectManager
 from spaceone.identity.manager.resource_manager import ResourceManager
 from spaceone.identity.manager.role_binding_manager import RoleBindingManager
-from spaceone.identity.manager.service_account_manager import ServiceAccountManager
-from spaceone.identity.manager.trusted_account_manager import TrustedAccountManager
-from spaceone.identity.manager.workspace_group_manager import WorkspaceGroupManager
+from spaceone.identity.manager.service_account_manager import \
+    ServiceAccountManager
+from spaceone.identity.manager.trusted_account_manager import \
+    TrustedAccountManager
+from spaceone.identity.manager.workspace_group_manager import \
+    WorkspaceGroupManager
 from spaceone.identity.manager.workspace_manager import WorkspaceManager
 from spaceone.identity.model import Workspace
 from spaceone.identity.model.workspace.request import *
@@ -36,6 +39,7 @@ def __init__(self, *args, **kwargs):
         self.resource_mgr = ResourceManager()
         self.workspace_mgr = WorkspaceManager()
         self.service_account_mgr = ServiceAccountManager()
+        self.workspace_group_mgr = WorkspaceGroupManager()
 
     @transaction(permission="identity:Workspace.write", role_types=["DOMAIN_ADMIN"])
     @convert_model
@@ -107,18 +111,40 @@ def change_workspace_group(
             workspace_id=params.workspace_id, domain_id=domain_id
         )
 
-        previous_workspace_group_id = workspace_vo.workspace_group_id
+        old_workspace_group_id = workspace_vo.workspace_group_id
         is_updatable = True
+        workspace_group_vo = None
         if workspace_group_id:
-            self._add_workspace_to_group(workspace_id, workspace_group_id, domain_id)
-        elif previous_workspace_group_id:
-            self._remove_workspace_from_group(previous_workspace_group_id, domain_id)
+            workspace_group_vo = self.workspace_group_mgr.get_workspace_group(
+                workspace_group_id, domain_id
+            )
+            is_updatable = self._add_workspace_to_group(
+                workspace_id, workspace_group_id, domain_id
+            )
+        elif old_workspace_group_id:
+            workspace_group_vo = self.workspace_group_mgr.get_workspace_group(
+                old_workspace_group_id, domain_id
+            )
+            self._remove_workspace_from_group(old_workspace_group_id, domain_id)
 
         if is_updatable:
             workspace_vo = self.workspace_mgr.update_workspace_by_vo(
                 params.dict(exclude_unset=False), workspace_vo
             )
 
+            workspace_vos = None
+            if workspace_group_id:
+                workspace_vos = self.workspace_mgr.filter_workspaces(
+                    workspace_group_id=workspace_group_id, domain_id=domain_id
+                )
+            elif old_workspace_group_id:
+                workspace_vos = self.workspace_mgr.filter_workspaces(
+                    workspace_group_id=old_workspace_group_id, domain_id=domain_id
+                )
+            self.workspace_group_mgr.update_workspace_group_by_vo(
+                {"workspace_count": len(workspace_vos)}, workspace_group_vo
+            )
+
         return WorkspaceResponse(**workspace_vo.to_dict())
 
     @transaction(permission="identity:Workspace.write", role_types=["DOMAIN_ADMIN"])
@@ -378,17 +404,16 @@ def _add_workspace_to_group(
         workspace_vo = self.workspace_mgr.get_workspace(
             workspace_id=workspace_id, domain_id=domain_id
         )
-        workspace_group_mgr = WorkspaceGroupManager()
-        existing_workspace_group_id = workspace_vo.workspace_group_id
+        old_workspace_group_id = workspace_vo.workspace_group_id
         is_updatable = True
 
-        workspace_group_vo = workspace_group_mgr.get_workspace_group(
+        workspace_group_vo = self.workspace_group_mgr.get_workspace_group(
             workspace_group_id=workspace_group_id, domain_id=domain_id
         )
 
-        if existing_workspace_group_id:
-            if existing_workspace_group_id != workspace_group_id:
-                self._delete_role_bindings(existing_workspace_group_id, domain_id)
+        if old_workspace_group_id:
+            if old_workspace_group_id != workspace_group_id:
+                self._delete_role_bindings(old_workspace_group_id, domain_id)
 
                 self._create_role_bindings(
                     workspace_group_vo.users,
@@ -409,9 +434,9 @@ def _add_workspace_to_group(
         return is_updatable
 
     def _remove_workspace_from_group(
-        self, previous_workspace_group_id: str, domain_id: str
+        self, old_workspace_group_id: str, domain_id: str
     ) -> None:
-        self._delete_role_bindings(previous_workspace_group_id, domain_id)
+        self._delete_role_bindings(old_workspace_group_id, domain_id)
 
     @staticmethod
     def _delete_role_bindings(existing_workspace_group_id: str, domain_id: str):