feat: add new feature WorkspaceConfig resource

src/spaceone/config/interface/grpc/__init__.py

@@ -1,9 +1,11 @@
 from spaceone.core.pygrpc.server import GRPCServer
 from spaceone.config.interface.grpc.domain_config import DomainConfig
+from spaceone.config.interface.grpc.workspace_config import WorkspaceConfig
 from spaceone.config.interface.grpc.user_config import UserConfig
 
 _all_ = ["app"]
 
 app = GRPCServer()
 app.add_service(DomainConfig)
+app.add_service(WorkspaceConfig)
 app.add_service(UserConfig)

src/spaceone/config/interface/grpc/workspace_config.py

@@ -0,0 +1,53 @@
+from spaceone.api.config.v1 import workspace_config_pb2, workspace_config_pb2_grpc
+from spaceone.core.pygrpc import BaseAPI
+
+from spaceone.config.service.workspace_config_service import WorkspaceConfigService
+from spaceone.config.info.domain_config_info import *
+from spaceone.config.info.common_info import *
+
+
+class WorkspaceConfig(BaseAPI, workspace_config_pb2_grpc.WorkspaceConfigServicer):
+    pb2 = workspace_config_pb2
+    pb2_grpc = workspace_config_pb2_grpc
+
+    def create(self, request, context):
+        params, metadata = self.parse_request(request, context)
+        workspace_config_svc = WorkspaceConfigService(metadata)
+        response: dict = workspace_config_svc.create(params)
+        return self.dict_to_message(response)
+
+    def update(self, request, context):
+        params, metadata = self.parse_request(request, context)
+        workspace_config_svc = WorkspaceConfigService(metadata)
+        response: dict = workspace_config_svc.update(params)
+        return self.dict_to_message(response)
+
+    def set(self, request, context):
+        params, metadata = self.parse_request(request, context)
+        workspace_config_svc = WorkspaceConfigService(metadata)
+        response: dict = workspace_config_svc.set(params)
+        return self.dict_to_message(response)
+
+    def delete(self, request, context):
+        params, metadata = self.parse_request(request, context)
+        workspace_config_svc = WorkspaceConfigService(metadata)
+        workspace_config_svc.delete(params)
+        return self.empty()
+
+    def get(self, request, context):
+        params, metadata = self.parse_request(request, context)
+        workspace_config_svc = WorkspaceConfigService(metadata)
+        response: dict = workspace_config_svc.get(params)
+        return self.dict_to_message(response)
+
+    def list(self, request, context):
+        params, metadata = self.parse_request(request, context)
+        workspace_config_svc = WorkspaceConfigService(metadata)
+        response: dict = workspace_config_svc.list(params)
+        return self.dict_to_message(response)
+
+    def stat(self, request, context):
+        params, metadata = self.parse_request(request, context)
+        workspace_config_svc = WorkspaceConfigService(metadata)
+        response: dict = workspace_config_svc.stat(params)
+        return self.dict_to_message(response)

src/spaceone/config/manager/workspace_config_manager.py

@@ -0,0 +1,56 @@
+import logging
+from typing import Tuple
+
+from mongoengine import QuerySet
+from spaceone.core.manager import BaseManager
+
+from spaceone.config.model.workspace_config.database import WorkspaceConfig
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class WorkspaceConfigManager(BaseManager):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.workspace_config_model = WorkspaceConfig
+
+    def create_workspace_config(self, params: dict) -> WorkspaceConfig:
+        def _rollback(vo: WorkspaceConfig) -> None:
+            _LOGGER.info(
+                f"[create_workspace_config._rollback] " f"Delete workspace config : {vo.name}"
+            )
+            vo.delete()
+
+        workspace_config_vo: WorkspaceConfig = self.workspace_config_model.create(params)
+        self.transaction.add_rollback(_rollback, workspace_config_vo)
+
+        return workspace_config_vo
+
+    def update_workspace_config_by_vo(
+            self, params: dict, workspace_config_vo: WorkspaceConfig
+    ) -> WorkspaceConfig:
+        def _rollback(old_data: dict):
+            _LOGGER.info(
+                f'[update_workspace_config_by_vo._rollback] Revert Data : {old_data["name"]}'
+            )
+            workspace_config_vo.update(old_data)
+
+        self.transaction.add_rollback(_rollback, workspace_config_vo.to_dict())
+
+        return workspace_config_vo.update(params)
+
+    @staticmethod
+    def delete_workspace_config_by_vo(workspace_config_vo: WorkspaceConfig) -> None:
+        workspace_config_vo.delete()
+
+    def get_workspace_config(self, name: str, workspace_id: str, domain_id: str) -> WorkspaceConfig:
+        return self.workspace_config_model.get(name=name, workspace_id=workspace_id, domain_id=domain_id)
+
+    def filter_workspace_configs(self, **conditions):
+        return self.workspace_config_model.filter(**conditions)
+
+    def list_workspace_configs(self, query: dict) -> Tuple[QuerySet, int]:
+        return self.workspace_config_model.query(**query)
+
+    def stat_workspace_configs(self, query: dict) -> dict:
+        return self.workspace_config_model.stat(**query)

src/spaceone/config/model/__init__.py

@@ -1,2 +1,3 @@
 from spaceone.config.model.user_config_model import UserConfig
+from spaceone.config.model.workspace_config.database import WorkspaceConfig
 from spaceone.config.model.domain_config_model import DomainConfig

src/spaceone/config/model/workspace_config/database.py

@@ -0,0 +1,24 @@
+from mongoengine import *
+
+from spaceone.core.model.mongo_model import MongoModel
+
+
+class WorkspaceConfig(MongoModel):
+    name = StringField(max_length=255, unique_with=["domain_id", "workspace_id"])
+    data = DictField()
+    tags = DictField()
+    workspace_id = StringField(max_length=40)
+    domain_id = StringField(max_length=40)
+    created_at = DateTimeField(auto_now_add=True)
+    updated_at = DateTimeField(auto_now=True)
+
+    meta = {
+        "updatable_fields": ["name", "data", "tags", "updated_at"],
+        "minimal_fields": ["name"],
+        "ordering": ["name"],
+        "indexes": [
+            "name",
+            "workspace_id",
+            "domain_id",
+        ],
+    }

src/spaceone/config/model/workspace_config/request.py

@@ -0,0 +1,61 @@
+from typing import Union
+from pydantic import BaseModel
+
+__all__ = [
+    "WorkspaceConfigCreateRequest",
+    "WorkspaceConfigUpdateRequest",
+    "WorkspaceConfigSetRequest",
+    "WorkspaceConfigDeleteRequest",
+    "WorkspaceConfigGetRequest",
+    "WorkspaceConfigSearchQueryRequest",
+    "WorkspaceConfigQueryRequest",
+]
+
+
+class WorkspaceConfigCreateRequest(BaseModel):
+    name: str
+    data: dict
+    tags: Union[dict, None] = None
+    workspace_id: str
+    domain_id: str
+
+
+class WorkspaceConfigUpdateRequest(BaseModel):
+    name: str
+    data: Union[dict, None] = None
+    tags: Union[dict, None] = None
+    workspace_id: str
+    domain_id: str
+
+
+class WorkspaceConfigSetRequest(BaseModel):
+    name: str
+    data: str
+    tags: Union[dict, None] = None
+    workspace_id: str
+    domain_id: str
+
+
+class WorkspaceConfigDeleteRequest(BaseModel):
+    name: str
+    workspace_id: str
+    domain_id: str
+
+
+class WorkspaceConfigGetRequest(BaseModel):
+    name: str
+    workspace_id: str
+    domain_id: str
+
+
+class WorkspaceConfigSearchQueryRequest(BaseModel):
+    query: Union[dict, None] = None
+    name: Union[str, None] = None
+    workspace_id: str
+    domain_id: str
+
+
+class WorkspaceConfigQueryRequest(BaseModel):
+    query: dict
+    workspace_id: str
+    domain_id: str

src/spaceone/config/model/workspace_config/response.py

@@ -0,0 +1,28 @@
+from datetime import datetime
+from typing import Union, List
+from pydantic import BaseModel
+
+from spaceone.core import utils
+
+__all__ = ["WorkspaceConfigResponse", "WorkspaceConfigsResponse"]
+
+
+class WorkspaceConfigResponse(BaseModel):
+    name: Union[str, None] = None
+    data: Union[dict, None] = None
+    tags: Union[dict, None] = None
+    workspace_id: Union[str, None] = None
+    domain_id: Union[str, None] = None
+    created_at: Union[datetime, None] = None
+    updated_at: Union[datetime, None] = None
+
+    def dict(self, *args, **kwargs):
+        data = super().dict(*args, **kwargs)
+        data["created_at"] = utils.datetime_to_iso8601(data["created_at"])
+        data["updated_at"] = utils.datetime_to_iso8601(data.get("updated_at"))
+        return data
+
+
+class WorkspaceConfigsResponse(BaseModel):
+    results: List[WorkspaceConfigResponse] = []
+    total_count: int