Merge pull request #5 from Sooyoung98/master

feat: modified dockerfile to use python-core 2.0 and applied new plug…

Dockerfile

@@ -1,4 +1,4 @@
-FROM cloudforet/python-core:1.12
+FROM cloudforet/python-core:2.0
 ARG PACKAGE_VERSION
 ENV PYTHONUNBUFFERED 1
 ENV SPACEONE_PORT 50051
@@ -21,4 +21,4 @@ RUN pip install --upgrade spaceone-api
 EXPOSE ${SPACEONE_PORT}
 
 ENTRYPOINT ["spaceone"]
-CMD ["grpc", "spaceone.notification", "-m", "/opt"]
+CMD ["run", "grpc-server", "spaceone.notification", "-m", "/opt"]

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

@@ -0,0 +1,17 @@
+from spaceone.core.pygrpc.server import GRPCServer
+from spaceone.notification.interface.grpc.protocol import Protocol
+from spaceone.notification.interface.grpc.user_channel import UserChannel
+from spaceone.notification.interface.grpc.project_channel import ProjectChannel
+from spaceone.notification.interface.grpc.notification import Notification
+from spaceone.notification.interface.grpc.notification_usage import NotificationUsage
+from spaceone.notification.interface.grpc.quota import Quota
+
+_all_ = ['app']
+
+app = GRPCServer()
+app.add_service(Protocol)
+app.add_service(UserChannel)
+app.add_service(ProjectChannel)
+app.add_service(Notification)
+app.add_service(NotificationUsage)
+app.add_service(Quota)

src/spaceone/notification/plugin/protocol/conf/global_conf.py

@@ -0,0 +1,15 @@
+LOG = {
+    'filters': {
+        'masking': {
+            'rules': {
+                'Protocol.verify': [
+                    'secret_data'
+                ],
+                'Notification.dispatch': [
+                    'secret_data',
+                    'channel_data'
+                ]
+            }
+        }
+    }
+}

src/spaceone/notification/plugin/protocol/interface/grpc/__init__.py

@@ -0,0 +1,9 @@
+from spaceone.core.pygrpc.server import GRPCServer
+from spaceone.notification.plugin.protocol.interface.grpc.protocol import Protocol
+from spaceone.notification.plugin.protocol.interface.grpc.notification import Notification
+
+_all_ = ['app']
+
+app = GRPCServer()
+app.add_service(Protocol)
+app.add_service(Notification)

src/spaceone/notification/plugin/protocol/interface/grpc/notification.py

@@ -0,0 +1,14 @@
+from spaceone.core.pygrpc import BaseAPI
+from spaceone.api.notification.plugin import notification_pb2, notification_pb2_grpc
+from spaceone.notification.plugin.protocol.service.notification_service import NotificationService
+
+
+class Notification(BaseAPI, notification_pb2_grpc.NotificationServicer):
+    pb2 = notification_pb2
+    pb2_grpc = notification_pb2_grpc
+
+    def dispatch(self, request, context):
+        params, metadata = self.parse_request(request, context)
+        notification_svc = NotificationService(metadata)
+        notification_svc.dispatch(params)
+        return self.empty()

src/spaceone/notification/plugin/protocol/interface/grpc/protocol.py

@@ -0,0 +1,20 @@
+from spaceone.core.pygrpc import BaseAPI
+from spaceone.api.notification.plugin import protocol_pb2, protocol_pb2_grpc
+from spaceone.notification.plugin.protocol.service.protocol_service import ProtocolService
+
+
+class Protocol(BaseAPI, protocol_pb2_grpc.ProtocolServicer):
+    pb2 = protocol_pb2
+    pb2_grpc = protocol_pb2_grpc
+
+    def init(self, request, context):
+        params, metadata = self.parse_request(request, context)
+        protocol_svc = ProtocolService(metadata)
+        response: dict = protocol_svc.init(params)
+        return self.dict_to_message(response)
+
+    def verify(self, request, context):
+        params, metadata = self.parse_request(request, context)
+        protocol_svc = ProtocolService(metadata)
+        protocol_svc.verify(params)
+        return self.empty()

src/spaceone/notification/plugin/protocol/lib/server.py

@@ -0,0 +1,22 @@
+from spaceone.core.pygrpc.server import GRPCServer
+from spaceone.core.plugin.server import PluginServer
+from spaceone.notification.plugin.protocol.interface.grpc import app
+from spaceone.notification.plugin.protocol.service.notification_service import NotificationService
+from spaceone.notification.plugin.protocol.service.protocol_service import ProtocolService
+
+__all__ = ['ProtocolPluginServer']
+
+
+class ProtocolPluginServer(PluginServer):
+    _grpc_app: GRPCServer = app
+    _global_conf_path: str = 'spaceone.notification.plugin.protocol.conf.global_conf:global_conf'
+    _plugin_methods = {
+        'Protocol': {
+            'service': ProtocolService,
+            'methods': ['init', 'verify']
+        },
+        'Notification': {
+            'service': NotificationService,
+            'methods': ['dispatch']
+        }
+    }

src/spaceone/notification/plugin/protocol/model/notification_request.py

@@ -0,0 +1,42 @@
+from typing import Union, Literal, List
+from pydantic import BaseModel
+
+__all__ = ['NotificationDispatchRequest']
+
+NotificationType = Literal['INFO', 'ERROR', 'SUCCESS', 'WARNING']
+ContentType = Literal['HTML', 'MARKDOWN']
+
+
+class Tag(BaseModel):
+    key: str
+    value: str
+    options: Union[dict, None] = None
+
+
+class Callback(BaseModel):
+    label: str
+    url: str
+    options: Union[dict, None] = None
+
+
+class Message(BaseModel):
+    title: str
+    link: Union[str, None] = None
+    description: Union[str, None] = None
+    short_description: Union[str, None] = None
+    contents: Union[str, None] = None
+    content_type: Union[ContentType, None] = None
+    image_url: Union[str, None] = None
+    tags: Union[List[Tag], None] = None
+    callbacks: Union[List[Callback], None] = None
+    occurred_at: Union[str, None] = None
+    domain_name: Union[str, None] = None
+
+
+class NotificationDispatchRequest(BaseModel):
+    options: dict
+    secret_data: dict
+    channel_data: dict
+    message: Message
+    notification_type: NotificationType
+    domain_id: Union[str, None] = None

src/spaceone/notification/plugin/protocol/model/protocol_request.py

@@ -0,0 +1,15 @@
+from typing import Union
+from pydantic import BaseModel
+
+__all__ = ['ProtocolInitRequest', 'ProtocolVerifyRequest']
+
+
+class ProtocolInitRequest(BaseModel):
+    options: dict
+    domain_id: Union[str, None] = None
+
+
+class ProtocolVerifyRequest(BaseModel):
+    options: dict
+    secret_data: dict
+    domain_id: Union[str, None] = None

src/spaceone/notification/plugin/protocol/model/protocol_response.py

@@ -0,0 +1,24 @@
+from typing import Literal
+from pydantic import BaseModel, Field
+
+__all__ = ['PluginResponse']
+
+DataType = Literal['PLAIN_TEXT', 'SECRET']
+
+
+class PluginDataSchema(BaseModel):
+    json_schema: dict = Field(alias="schema")
+
+    def dict(self, *args, **kwargs):
+        data = super().dict(*args, **kwargs)
+        data['schema'] = data.pop('json_schema')
+        return data
+
+
+class PluginMetadata(BaseModel):
+    data_type: DataType
+    data: PluginDataSchema
+
+
+class PluginResponse(BaseModel):
+    metadata: PluginMetadata

src/spaceone/notification/plugin/protocol/service/notification_service.py

@@ -0,0 +1,28 @@
+import logging
+from spaceone.core.service import BaseService, transaction, convert_model
+from spaceone.notification.plugin.protocol.model.notification_request import NotificationDispatchRequest
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class NotificationService(BaseService):
+    @transaction
+    @convert_model
+    def dispatch(self, params: NotificationDispatchRequest) -> None:
+        """ dispatch notification
+
+        Args:
+            params (NotificationDispatchRequest): {
+                'options': 'dict',                  # Required
+                'secret_data': 'dict',              # Required
+                'channel_data': 'dict',             # Required
+                'message': 'dict',                  # Required
+                'notification_type': 'str',         # Required
+            }
+
+        Returns:
+            None
+        """
+
+        func = self.get_plugin_method('dispatch')
+        func(params.dict())

src/spaceone/notification/plugin/protocol/service/protocol_service.py

@@ -0,0 +1,48 @@
+import logging
+from typing import Generator, Union
+from spaceone.core.service import BaseService, transaction, convert_model
+from spaceone.notification.plugin.protocol.model.protocol_request import ProtocolInitRequest, ProtocolVerifyRequest
+from spaceone.notification.plugin.protocol.model.protocol_response import PluginResponse
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class ProtocolService(BaseService):
+
+    @transaction
+    @convert_model
+    def init(self, params: ProtocolInitRequest) -> Union[dict, PluginResponse]:
+        """ init plugin by options
+
+        Args:
+            params (ProtocolInitRequest): {
+                'options': 'dict',    # Required
+                'domain_id': 'str'
+            }
+
+        Returns:
+            PluginResponse:
+        """
+
+        func = self.get_plugin_method('init')
+        response = func(params.dict())
+        return PluginResponse(**response)
+
+    @transaction
+    @convert_model
+    def verify(self, params: ProtocolVerifyRequest) -> None:
+        """ Verifying protocol plugin
+
+        Args:
+            params (ProtocolVerifyRequest): {
+                'options': 'dict',      # Required
+                'secret_data': 'dict',  # Required
+                'domain_id': 'str'
+            }
+
+        Returns:
+            None
+        """
+
+        func = self.get_plugin_method('verify')
+        func(params.dict())

src/spaceone/notification/plugin/protocol/skeleton/main.py

@@ -0,0 +1,57 @@
+from spaceone.notification.plugin.protocol.lib.server import ProtocolPluginServer
+
+app = ProtocolPluginServer()
+
+
+@app.route('Protocol.init')
+def protocol_init(params: dict) -> dict:
+    """ init plugin by options
+
+    Args:
+        params (ProtocolInitRequest): {
+            'options': 'dict',    # Required
+            'domain_id': 'str'
+        }
+
+    Returns:
+        PluginResponse: {
+            'metadata': 'dict'
+        }
+    """
+    pass
+
+
+@app.route('Protocol.verify')
+def protocol_verify(params: dict) -> None:
+    """ Verifying protocol plugin
+
+    Args:
+        params (ProtocolVerifyRequest): {
+            'options': 'dict',      # Required
+            'secret_data': 'dict',  # Required
+            'domain_id': 'str'
+        }
+
+    Returns:
+        None
+    """
+    pass
+
+
+@app.route('Notification.dispatch')
+def notification_dispatch(params: dict) -> None:
+    """ dispatch notification
+
+    Args:
+        params (NotificationDispatchRequest): {
+            'options': 'dict',              # Required
+            'secret_data': 'dict',          # Required
+            'channel_data': 'dict',         # Required
+            'message': 'dict',              # Required
+            'notification_type': 'str',     # Required
+        }
+
+    Returns:
+        None
+    """
+    pass