Add change_workspace_group api

src/spaceone/identity/interface/grpc/workspace.py

@@ -1,5 +1,6 @@
-from spaceone.core.pygrpc import BaseAPI
 from spaceone.api.identity.v2 import workspace_pb2, workspace_pb2_grpc
+from spaceone.core.pygrpc import BaseAPI
+
 from spaceone.identity.service.workspace_service import WorkspaceService
 
 
@@ -19,6 +20,12 @@ def update(self, request, context):
         response: dict = workspace_svc.update(params)
         return self.dict_to_message(response)
 
+    def change_workspace_group(self, request, context):
+        params, metadata = self.parse_request(request, context)
+        workspace_svc = WorkspaceService(metadata)
+        response: dict = workspace_svc.change_workspace_group(params)
+        return self.dict_to_message(response)
+
     def delete(self, request, context):
         params, metadata = self.parse_request(request, context)
         workspace_svc = WorkspaceService(metadata)

src/spaceone/identity/manager/workspace_manager.py

@@ -60,7 +60,11 @@ def delete_workspace_by_vo(workspace_vo: Workspace) -> None:
             _LOGGER.debug(
                 f"[delete_workspace_by_vo] Delete role bindings count with {workspace_vo.workspace_id} : {rb_vos.count()}"
             )
-            rb_vos.delete()
+            for rb_vo in rb_vos:
+                _LOGGER.debug(
+                    f"[delete_role_binding_by_vo] Delete role binding info: {rb_vo.to_dict()}"
+                )
+                rb_vo.delete()
 
         workspace_vo.delete()
 

src/spaceone/identity/model/workspace/request.py

@@ -1,9 +1,11 @@
-from typing import Union, Literal
+from typing import Literal, Union
+
 from pydantic import BaseModel
 
 __all__ = [
     "WorkspaceCreateRequest",
     "WorkspaceUpdateRequest",
+    "WorkspaceChangeWorkspaceGroupRequest",
     "WorkspaceDeleteRequest",
     "WorkspaceEnableRequest",
     "WorkspaceDisableRequest",
@@ -30,6 +32,12 @@ class WorkspaceUpdateRequest(BaseModel):
     domain_id: str
 
 
+class WorkspaceChangeWorkspaceGroupRequest(BaseModel):
+    workspace_id: str
+    workspace_group_id: Union[str, None] = None
+    domain_id: str
+
+
 class WorkspaceDeleteRequest(BaseModel):
     force: Union[bool, None] = False
     workspace_id: str
@@ -64,6 +72,7 @@ class WorkspaceSearchQueryRequest(BaseModel):
     created_by: Union[str, None] = None
     is_managed: Union[bool, None] = None
     is_dormant: Union[bool, None] = None
+    workspace_group_id: Union[str, None] = None
     domain_id: str
 
 

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

@@ -23,6 +23,7 @@ class WorkspaceResponse(BaseModel):
     user_count: Union[int, None] = None
     cost_info: Union[dict, None] = None
     trusted_account_id: Union[str, None] = None
+    workspace_group_id: Union[str, None] = None
     domain_id: Union[str, None] = None
     created_at: Union[datetime, None] = None
     last_synced_at: Union[datetime, None] = None

src/spaceone/identity/service/workspace_service.py

@@ -1,5 +1,5 @@
 import logging
-from typing import Union
+from typing import Dict, List, Union
 
 from spaceone.core.error import *
 from spaceone.core.service import *
@@ -11,16 +11,14 @@
 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 *
 from spaceone.identity.model.workspace.response import *
+from spaceone.identity.service.role_binding_service import RoleBindingService
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -86,6 +84,45 @@ def update(self, params: WorkspaceUpdateRequest) -> Union[WorkspaceResponse, dic
         )
         return WorkspaceResponse(**workspace_vo.to_dict())
 
+    @transaction(permission="identity:Workspace.write", role_types=["DOMAIN_ADMIN"])
+    @convert_model
+    def change_workspace_group(
+        self, params: WorkspaceChangeWorkspaceGroupRequest
+    ) -> Union[WorkspaceResponse, dict]:
+        """Change workspace group
+        Args:
+            params (WorkspaceChangeWorkspaceGroupRequest): {
+                'workspace_id': 'str',                   # required
+                'workspace_group_id': 'str',
+                'domain_id': 'str'                       # injected from auth (required)
+            }
+        Returns:
+            WorkspaceResponse:
+        """
+        workspace_id = params.workspace_id
+        workspace_group_id = params.workspace_group_id
+        domain_id = params.domain_id
+
+        workspace_vo = self.workspace_mgr.get_workspace(
+            workspace_id=params.workspace_id, domain_id=domain_id
+        )
+
+        previous_workspace_group_id = workspace_vo.workspace_group_id
+        is_updatable = True
+        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, workspace_group_id, domain_id
+            )
+
+        if is_updatable:
+            workspace_vo = self.workspace_mgr.update_workspace_by_vo(
+                params.dict(exclude_unset=False), workspace_vo
+            )
+
+        return WorkspaceResponse(**workspace_vo.to_dict())
+
     @transaction(permission="identity:Workspace.write", role_types=["DOMAIN_ADMIN"])
     @convert_model
     def delete(self, params: WorkspaceDeleteRequest) -> None:
@@ -230,6 +267,7 @@ def list(
                 'created_by': 'str',
                 'is_managed': 'bool',
                 'is_dormant': 'bool',
+                'workspace_group_id': 'str',
                 'domain_id': 'str',         # injected from auth (required)
             }
         Returns:
@@ -335,3 +373,77 @@ def _delete_related_resources_in_workspace(workspace_vo: Workspace):
             _LOGGER.debug(
                 f"[_delete_related_resources_in_workspace] Delete workspace group: {workspace_group_vo.name} ({workspace_group_vo.workspace_group_id})"
             )
+
+    def _add_workspace_to_group(
+        self, workspace_id: str, workspace_group_id: str, domain_id: str
+    ) -> bool:
+        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
+        is_updatable = True
+
+        workspace_group_vo = 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)
+
+                self._create_role_bindings(
+                    workspace_group_vo.users,
+                    workspace_id,
+                    workspace_group_id,
+                    domain_id,
+                )
+            else:
+                is_updatable = False
+        else:
+            self._create_role_bindings(
+                workspace_group_vo.users,
+                workspace_id,
+                workspace_group_id,
+                domain_id,
+            )
+
+        return is_updatable
+
+    def _remove_workspace_from_group(
+        self, previous_workspace_group_id: str, workspace_group_id: str, domain_id: str
+    ) -> None:
+        self._delete_role_bindings(previous_workspace_group_id, domain_id)
+
+    @staticmethod
+    def _delete_role_bindings(existing_workspace_group_id: str, domain_id: str):
+        rb_mgr = RoleBindingManager()
+        rb_vos = rb_mgr.filter_role_bindings(
+            workspace_group_id=existing_workspace_group_id,
+            domain_id=domain_id,
+        )
+        for rb_vo in rb_vos:
+            _LOGGER.debug(
+                f"[_delete_role_bindings] Delete role binding info: {rb_vo.to_dict()}"
+            )
+            rb_vo.delete()
+
+    @staticmethod
+    def _create_role_bindings(
+        workspace_group_users: List[Dict[str, str]],
+        workspace_id: str,
+        workspace_group_id: str,
+        domain_id: str,
+    ):
+        rb_svc = RoleBindingService()
+        for user_info in workspace_group_users or []:
+            rb_svc.create_role_binding(
+                {
+                    "user_id": user_info["user_id"],
+                    "role_id": user_info["role_id"],
+                    "resource_group": "WORKSPACE",
+                    "domain_id": domain_id,
+                    "workspace_group_id": workspace_group_id,
+                    "workspace_id": workspace_id,
+                }
+            )