[amg] Azure managed grafana options for notification channels

src/amg/README.md

@@ -26,7 +26,7 @@ az grafana delete \
     -n MyGrafanaInstance
 ```
 
-### Configure folder, data sources and dashboard
+### Configure folders, data sources, notification channels and dashboards
 
 #### create a folder
 *Examples:*
@@ -44,6 +44,14 @@ az grafana data-source create \
     --definition ~/data-source-sql.json
 ```
 
+#### configure a notification channel 
+*Examples:*
+```
+az grafana notification-channel create \
+    -n MyGrafanaInstance \
+    --definition ~/notification-channel-teams.json
+```
+
 #### Create a dashboard
 *Examples:*
 ```

src/amg/azext_amg/_help.py

@@ -61,7 +61,6 @@
             }'
 """
 
-
 helps['grafana data-source update'] = """
     type: command
     short-summary: Update a data source.
@@ -87,6 +86,57 @@
     short-summary: query a data source having backend implementation
 """
 
+helps['grafana notification-channel'] = """
+    type: group
+    short-summary: Commands to manage notification channels of an instance.
+"""
+
+helps['grafana notification-channel list'] = """
+    type: command
+    short-summary: List all notification channels of an instance.
+"""
+
+helps['grafana notification-channel list-short'] = """
+    type: command
+    short-summary: List all notification channels of an instance in short.
+"""
+
+helps['grafana notification-channel show'] = """
+    type: command
+    short-summary: get details of a notification channel
+"""
+
+helps['grafana notification-channel create'] = """
+    type: command
+    short-summary: Create a notification channel.
+    examples:
+        - name: create a notification channel for Teams
+          text: |
+            az grafana notification-channel create -n MyGrafana --definition '{
+              "name": "Teams",
+              "settings": {
+                "uploadImage": true,
+                "url": "https://webhook.office.com/IncomingWebhook/"
+               },
+              "type": "teams"
+            }'
+"""
+
+helps['grafana notification-channel update'] = """
+    type: command
+    short-summary: Update a notification channel.
+"""
+
+helps['grafana notification-channel delete'] = """
+    type: command
+    short-summary: delete a notification channel.
+"""
+
+helps['grafana notification-channel test'] = """
+    type: command
+    short-summary: tests a notification channels.
+"""
+
 helps['grafana dashboard'] = """
     type: group
     short-summary: Commands to manage dashboards of an instance.

src/amg/azext_amg/_params.py

@@ -43,6 +43,10 @@ def load_arguments(self, _):
         c.argument("data_source", help="name, id, uid which can identify a data source. CLI will search in the order of name, id, and uid, till finds a match")
         c.argument("definition", help="json string with data source definition, or a path to a file with such content")
 
+    with self.argument_context("grafana notification-channel") as c:
+        c.argument("notification_channel", help="id, uid which can identify a data source. CLI will search in the order of id, and uid, till finds a match")
+        c.argument("definition", help="json string with notification channel definition, or a path to a file with such content")
+
     with self.argument_context("grafana data-source query") as c:
         c.argument("conditions", nargs="+", help="space-separated condition in a format of `<name>=<value>`")
         c.argument("time_from", options_list=["--from"], help="start time in iso 8601, e.g. '2022-01-02T16:15:00'. Default: 1 hour early")

src/amg/azext_amg/commands.py

@@ -32,6 +32,15 @@ def load_command_table(self, _):
         g.custom_command('query', 'query_data_source')
         g.custom_command('update', 'update_data_source')
 
+    with self.command_group('grafana notification-channel') as g:
+        g.custom_command('list', 'list_notification_channels')
+        g.custom_command('list-short', 'list_notification_channels_short')
+        g.custom_show_command('show', 'show_notification_channel')
+        g.custom_command('create', 'create_notification_channel')
+        g.custom_command('update', 'update_notification_channel')
+        g.custom_command('delete', 'delete_notification_channel')
+        g.custom_command('test', 'test_notification_channel')
+
     with self.command_group('grafana folder') as g:
         g.custom_command('create', 'create_folder')
         g.custom_command('list', 'list_folders')

src/amg/azext_amg/custom.py

@@ -283,6 +283,49 @@ def update_data_source(cmd, grafana_name, data_source, definition, resource_grou
     return json.loads(response.content)
 
 
+def list_notification_channels(cmd, grafana_name, resource_group_name=None):
+    response = _send_request(cmd, resource_group_name, grafana_name, "get", "/api/alert-notifications")
+    return json.loads(response.content)
+
+
+def list_notification_channels_short(cmd, grafana_name, resource_group_name=None):
+    response = _send_request(cmd, resource_group_name, grafana_name, "get", "/api/alert-notifications/lookup")
+    return json.loads(response.content)
+
+
+def show_notification_channel(cmd, grafana_name, notification_channel, resource_group_name=None):
+    return _find_notification_channel(cmd, resource_group_name, grafana_name, notification_channel)
+
+
+def create_notification_channel(cmd, grafana_name, definition, resource_group_name=None):
+    definition = _try_load_file_content(definition)
+    payload = json.loads(definition)
+    response = _send_request(cmd, resource_group_name, grafana_name, "post", "/api/alert-notifications", payload)
+    return json.loads(response.content)
+
+
+def update_notification_channel(cmd, grafana_name, notification_channel, definition, resource_group_name=None):
+    definition = json.loads(_try_load_file_content(definition))
+    data = _find_notification_channel(cmd, resource_group_name, grafana_name, notification_channel)
+    definition['id'] = data['id']
+    response = _send_request(cmd, resource_group_name, grafana_name, "put",
+                             "/api/alert-notifications/" + str(data['id']),
+                             definition)
+    return json.loads(response.content)
+
+
+def delete_notification_channel(cmd, grafana_name, notification_channel, resource_group_name=None):
+    data = _find_notification_channel(cmd, resource_group_name, grafana_name, notification_channel)
+    _send_request(cmd, resource_group_name, grafana_name, "delete", "/api/alert-notifications/" + str(data["id"]))
+
+
+def test_notification_channel(cmd, grafana_name, notification_channel, resource_group_name=None):
+    data = _find_notification_channel(cmd, resource_group_name, grafana_name, notification_channel)
+    response = _send_request(cmd, resource_group_name, grafana_name, "post", "/api/alert-notifications/test",
+                             data)
+    return json.loads(response.content)
+
+
 def create_folder(cmd, grafana_name, title, resource_group_name=None):
     payload = {
         "title": title
@@ -415,6 +458,20 @@ def _find_data_source(cmd, resource_group_name, grafana_name, data_source):
     return json.loads(response.content)
 
 
+def _find_notification_channel(cmd, resource_group_name, grafana_name, notification_channel):
+    response = _send_request(cmd, resource_group_name, grafana_name, "get",
+                             "/api/alert-notifications/" + notification_channel,
+                             raise_for_error_status=False)
+    if response.status_code >= 400:
+        response = _send_request(cmd, resource_group_name, grafana_name,
+                                 "get", "/api/alert-notifications/uid/" + notification_channel,
+                                 raise_for_error_status=False)
+    if response.status_code >= 400:
+        raise ArgumentUsageError(
+            f"Couldn't found notification channel {notification_channel}. Ex: {response.status_code}")
+    return json.loads(response.content)
+
+
 # For UX: we accept a file path for complex payload such as dashboard/data-source definition
 def _try_load_file_content(file_content):
     import os

src/amg/setup.py

@@ -16,7 +16,7 @@
 
 # TODO: Confirm this is the right version number you want and it matches your
 # HISTORY.rst entry.
-VERSION = '0.1.1'
+VERSION = '0.1.2'
 
 # The full list of classifiers is available at
 # https://pypi.python.org/pypi?%3Aaction=list_classifiers