feat: dingTalk Robot message support

notifiers/providers/__init__.py

@@ -1,3 +1,4 @@
+from . import dingtalk
 from . import email
 from . import gitter
 from . import gmail
@@ -21,6 +22,7 @@
     "simplepush": simplepush.SimplePush,
     "slack": slack.Slack,
     "email": email.SMTP,
+    "dingtalk": dingtalk.DingTalk,
     "gmail": gmail.Gmail,
     "icloud": icloud.iCloud,
     "telegram": telegram.Telegram,

notifiers/providers/dingtalk.py

@@ -0,0 +1,91 @@
+from ..core import Provider
+from ..core import Response
+from ..utils import requests
+from ..utils.helpers import snake_to_camel_case
+
+
+class DingTalk(Provider):
+    """Send DingTalk notifications"""
+
+    base_url = "https://oapi.dingtalk.com/robot/send?access_token={}"
+    site_url = "https://oapi.dingtalk.com/"
+    path_to_errors = ("message",)
+    name = "dingtalk"
+
+    _required = {"required": ["access_token", "message"]}
+    _schema = {
+        "type": "object",
+        "properties": {
+            "access_token": {
+                "type": "string",
+                "title": "access token to pair a channel to receive notification",
+            },
+            "msg_type": {
+                "type": "string",
+                "title": "choose a message type, these type are supported: text, markdown, link, action_card",
+                "enum": ["text", "markdown", "link", "action_card"],
+            },
+            "message": {
+                "type": "string",
+                "title": "This is the text that will be posted to the dingtalk",
+                "maxLength": 4096,
+            },
+            "msg_title": {
+                "type": "string",
+                "title": "title for markdown message and card message",
+            },
+            "msg_url": {
+                "type": "string",
+                "format": "uri",
+                "title": "url for markdown message",
+            },
+            "msg_btn_title": {
+                "type": "string",
+                "title": "title for card message button, like 'Read more.'",
+            },
+            "msg_btn_url": {
+                "type": "string",
+                "format": "uri",
+                "title": "url for card message button",
+            },
+        },
+        "additionalProperties": False,
+    }
+
+    @property
+    def defaults(self) -> dict:
+        return {"msg_type": "text"}
+
+    def _prepare_data(self, data: dict) -> dict:
+        text = data.pop("message")
+        mapping_key = {
+            "msg_title": "title",
+            "msg_url": "messageUrl",
+            "msg_btn_title": "singleTitle",
+            "msg_btn_url": "singleURL",
+        }
+
+        new_data = {
+            "access_token": data.pop("access_token"),
+            "msgtype": data.pop("msg_type"),
+        }
+
+        if new_data["msgtype"] != "text":
+            camel_case_key = snake_to_camel_case(new_data["msgtype"])
+            new_data["msgtype"] = camel_case_key[0].lower() + camel_case_key[1:]
+
+            new_data[new_data["msgtype"]] = {"text": text}
+            for key in data:
+                new_data[new_data["msgtype"]][mapping_key[key]] = data[key]
+        else:
+            new_data["text"] = {"content": text}
+
+        return new_data
+
+    def _send_notification(self, data: dict) -> Response:
+        access_token = data.pop("access_token")
+        url = self.base_url.format(access_token)
+        response, errors = requests.post(
+            url, json=data, path_to_errors=self.path_to_errors
+        )
+        return self.create_response(data, response, errors)

source/providers/dingtalk.rst

@@ -0,0 +1,29 @@
+DingTalk
+----------
+Send `DingTalk Robot <https://dingtalk.com/>`_ notifications
+
+Minimal example:
+
+.. code-block:: python
+
+    >>> from notifiers import get_notifier
+    >>> dingtalk = get_notifier('dingtalk')
+    >>> dingtalk.notify(access_token='token', msg_data={'msgtype': 'text', 'text':{'content': 'Hi there!'}})
+
+Full schema:
+
+.. code-block:: yaml
+
+    additionalProperties: false
+    properties:
+      access_token:
+        title: your access token
+        type: string
+      msg_data:
+        title: your message definition
+        type: object
+    required:
+    - access_token
+    - msg_data
+    type: object
+

tests/providers/test_dingtalk.py

@@ -0,0 +1,18 @@
+import pytest
+
+provider = "dingtalk"
+
+
+class TestDingTalk:
+    def test_dingtalk_metadata(self, provider):
+        assert provider.metadata == {
+            "base_url": "https://oapi.dingtalk.com/robot/send?access_token={}",
+            "name": "dingtalk",
+            "site_url": "https://oapi.dingtalk.com/",
+        }
+
+    @pytest.mark.online
+    def test_sanity(self, provider, test_message):
+        msg_data = {"msgtype": "text", "text": {"content": test_message}}
+        data = {"access_token": "token", "msg_data": msg_data}
+        provider.notify(**data, raise_on_errors=True)