Integrating the notifications to Zoom Webhook

CONTRIBUTORS.md

@@ -13,6 +13,7 @@ Thanks to the contributors who helped on this project apart from the authors
 * [Jagadapi Sivanaga Krishnam Raja Reddy](www.linkedin.com/in/jskrajareddy/)
 * [Vigneshwarr Venkatesan](https://www.linkedin.com/in/vignesh15)
 * [Nishant Singh](https://www.linkedin.com/in/singh-nishant/)
+* [Amaldev Kunnel](https://www.linkedin.com/in/amaldev-k-40222680)
 
 # Honorary Mentions
 Thanks to the team below for invaluable insights and support throughout the initial release of this project

docs/api/zoom_plugin.md

@@ -0,0 +1,11 @@
+---
+search:
+ exclude: true
+---
+
+::: spark_expectations.notifications.plugins.zoom
+ handler: python
+ options:
+ filters:
+ - "!^_[^_]"
+ - "!^__[^__]"

spark_expectations/config/user_config.py

@@ -29,6 +29,13 @@ class Constants:
  "spark.expectations.notifications.teams.webhook_url"
  )
 
+ # declare const user config variables for zoom notification
+ se_notifications_enable_zoom = "spark.expectations.notifications.zoom.enabled"
+ se_notifications_zoom_webhook_url = (
+ "spark.expectations.notifications.zoom.webhook_url"
+ )
+ se_notifications_zoom_token = "spark.expectations.notifications.zoom.token"
+
  se_notifications_on_start = "spark.expectations.notifications.on_start"
  se_notifications_on_completion = "spark.expectations.notifications.on.completion"
  se_notifications_on_fail = "spark.expectations.notifications.on.fail"

spark_expectations/core/context.py

@@ -59,6 +59,10 @@ def __post_init__(self) -> None:
  self._enable_teams: bool = False
  self._teams_webhook_url: Optional[str] = None
 
+ self._enable_zoom: bool = False
+ self._zoom_webhook_url: Optional[str] = None
+ self._zoom_token: Optional[str] = None
+
  self._table_name: Optional[str] = None
  self._input_count: int = 0
  self._error_count: int = 0
@@ -622,6 +626,74 @@ def get_teams_webhook_url(self) -> str:
  accessing it"""
  )
 
+ # def set_enable_zoom(self, enable_zoom: bool, zoom_token: str) -> None:
+ def set_enable_zoom(self, enable_zoom: bool) -> None:
+ """
+ Set whether to enable Zoom notification and its token.
+
+ Args:
+ enable_zoom (bool): Whether to enable Zoom notification or not.
+ """
+ self._enable_zoom = enable_zoom
+
+ @property
+ def get_enable_zoom(self) -> bool:
+ """
+ Get whether Zoom notification is enabled.
+
+ Returns:
+ bool: Whether Zoom notification is enabled or not.
+ """
+ return self._enable_zoom
+
+ def set_zoom_webhook_url(self, zoom_webhook_url: str) -> None:
+ """
+ Set the Zoom webhook URL.
+
+ Args:
+ zoom_webhook_url (str): The webhook URL for Zoom notification.
+ """
+ self._zoom_webhook_url = zoom_webhook_url
+
+ @property
+ def get_zoom_webhook_url(self) -> str:
+ """
+ Get the Zoom webhook URL.
+
+ Returns:
+ str: The Zoom webhook URL.
+ """
+ if self._zoom_webhook_url:
+ return self._zoom_webhook_url
+ raise SparkExpectationsMiscException(
+ """The spark expectations context is not set completely, please assign '_zoom_webhook_url' before 
+ accessing it"""
+ )
+
+ def set_zoom_token(self, zoom_token: str) -> None:
+ """
+ Set the Zoom webhook token.
+
+ Args:
+ zoom_token (str): The token for Zoom notification.
+ """
+ self._zoom_token = zoom_token
+
+ @property
+ def get_zoom_token(self) -> str:
+ """
+ Get the Zoom token.
+
+ Returns:
+ str: The Zoom token.
+ """
+ if self._zoom_token:
+ return self._zoom_token
+ raise SparkExpectationsMiscException(
+ """The spark expectations context is not set completely, please assign '_zoom_token' before 
+ accessing it"""
+ )
+
  def set_table_name(self, table_name: str) -> None:
  self._table_name = table_name
 

spark_expectations/core/exceptions.py

@@ -57,6 +57,14 @@ class SparkExpectationsTeamsNotificationException(Exception):
  pass
 
 
+class SparkExpectationsZoomNotificationException(Exception):
+ """
+ Throw this exception when spark expectations encounters exceptions while sending Zoom notifications
+ """
+
+ pass
+
+
 class SparkExpectationsEmailException(Exception):
  """
  Throw this exception when spark expectations encounters exceptions while sending email notifications

spark_expectations/notifications/__init__.py

@@ -15,12 +15,15 @@
 from spark_expectations.notifications.plugins.teams import (
  SparkExpectationsTeamsPluginImpl,
 )
+from spark_expectations.notifications.plugins.zoom import (
+ SparkExpectationsZoomPluginImpl, # Import Zoom plugin
+)
 
 
 @functools.lru_cache
 def get_notifications_hook() -> pluggy.PluginManager:
  """
- function provides pluggy hook manger to send email and slack notification
+ function provides pluggy hook manger to send email, slack and zoom notification
  Returns:
  PluginManager: pluggy Manager object
 
@@ -36,6 +39,10 @@ def get_notifications_hook() -> pluggy.PluginManager:
  pm.register(
  SparkExpectationsTeamsPluginImpl(), "spark_expectations_teams_notification"
  )
+ pm.register(
+ SparkExpectationsZoomPluginImpl(),
+ "spark_expectations_zoom_notification", # Register Zoom plugin
+ )
  for name, plugin_instance in pm.list_name_plugin():
  _log.info(
  "Loaded plugin with name: %s and class: %s",

spark_expectations/notifications/plugins/zoom.py

@@ -0,0 +1,68 @@
+from typing import Dict, Union
+import requests
+from spark_expectations import _log
+from spark_expectations.notifications.plugins.base_notification import (
+ SparkExpectationsNotification,
+ spark_expectations_notification_impl,
+)
+from spark_expectations.core.exceptions import (
+ SparkExpectationsZoomNotificationException,
+)
+from spark_expectations.core.context import SparkExpectationsContext
+
+
+class SparkExpectationsZoomPluginImpl(SparkExpectationsNotification):
+ """
+ This class implements/supports functionality to send Zoom notification
+ """
+
+ @spark_expectations_notification_impl
+ def send_notification(
+ self,
+ _context: SparkExpectationsContext,
+ _config_args: Dict[str, Union[str, bool]],
+ ) -> None:
+ """
+ function to send the Zoom notification
+ Args:
+ _context: SparkExpectationsContext class object
+ _config_args: dict
+
+ Returns: None
+
+ """
+ try:
+ if _context.get_enable_zoom is True:
+ message = _config_args.get("message")
+
+ # Format Message for Zoom
+ if isinstance(message, str):
+ message = message.replace("\n", "\n\n").replace(" ", "")
+
+ payload = {
+ "title": "SE Notification",
+ "themeColor": "008000",
+ "text": message,
+ }
+ headers = {
+ "Authorization": f"Bearer {_context.get_zoom_token}", # Use get_zoom_token to retrieve token.
+ "Content-Type": "application/json",
+ }
+ response = requests.post(
+ _context.get_zoom_webhook_url,
+ json=payload,
+ headers=headers,
+ timeout=10,
+ )
+
+ # Check the response for success or failure
+ if response.status_code == 200:
+ _log.info("Message posted successfully!")
+ else:
+ _log.info("Failed to post message")
+ raise SparkExpectationsZoomNotificationException(
+ "error occurred while sending Zoom notification from spark expectations project"
+ )
+
+ except Exception as e:
+ raise SparkExpectationsZoomNotificationException(e)

spark_expectations/utils/reader.py

@@ -44,6 +44,9 @@ def set_notification_param(
  user_config.se_notifications_slack_webhook_url: "",
  user_config.se_notifications_enable_teams: False,
  user_config.se_notifications_teams_webhook_url: "",
+ user_config.se_notifications_enable_zoom: False,
+ user_config.se_notifications_zoom_webhook_url: "",
+ user_config.se_notifications_zoom_token: "",
  }
 
  _notification_dict: Dict[str, Union[str, int, bool]] = (
@@ -132,6 +135,30 @@ def set_notification_param(
  "All params/variables required for slack notification is not configured or supplied"
  )
 
+ if _notification_dict[user_config.se_notifications_enable_zoom] is True:
+ if _notification_dict[
+ user_config.se_notifications_zoom_webhook_url
+ ]:
+ self._context.set_enable_zoom(True)
+ self._context.set_zoom_webhook_url(
+ str(
+ _notification_dict[
+ user_config.se_notifications_zoom_webhook_url
+ ]
+ )
+ )
+ self._context.set_zoom_token(
+ str(
+ _notification_dict[
+ user_config.se_notifications_zoom_token
+ ]
+ )
+ )
+ else:
+ raise SparkExpectationsMiscException(
+ "All params/variables required for zoom notification is not configured or supplied"
+ )
+
  except Exception as e:
  raise SparkExpectationsMiscException(
  f"error occurred while reading notification configurations {e}"

tests/config/test_user_config.py

@@ -15,6 +15,12 @@ def test_constants():
 
  assert user_config.se_notifications_slack_webhook_url == "spark.expectations.notifications.slack.webhook_url"
 
+ assert user_config.se_notifications_enable_zoom == "spark.expectations.notifications.zoom.enabled"
+
+ assert user_config.se_notifications_zoom_webhook_url == "spark.expectations.notifications.zoom.webhook_url"
+
+ assert user_config.se_notifications_zoom_token == "spark.expectations.notifications.zoom.token"
+
  assert user_config.se_notifications_on_start == "spark.expectations.notifications.on_start"
 
  assert user_config.se_notifications_on_completion == "spark.expectations.notifications.on.completion"

tests/core/test_context.py

@@ -59,6 +59,9 @@ def test_context_properties():
  context._slack_webhook_url = "abcedfghi"
  context._enable_teams = True
  context._teams_webhook_url = "abcedfghi"
+ context._enable_zoom = True
+ context._zoom_webhook_url = "abcedfghi"
+ context._zoom_token = "abcedfghi"
  context._table_name = "test_table"
  context._input_count = 100
  context._error_count = 10
@@ -173,6 +176,9 @@ def test_context_properties():
  assert context._slack_webhook_url == "abcedfghi"
  assert context._enable_teams is True
  assert context._teams_webhook_url == "abcedfghi"
+ assert context._enable_zoom is True
+ assert context._zoom_webhook_url == "abcedfghi"
+ assert context._zoom_token == "abcedfghi"
  assert context._table_name == "test_table"
  assert context._input_count == 100
  assert context._error_count == 10
@@ -535,6 +541,27 @@ def test_set_teams_webhook_url():
  assert context.get_teams_webhook_url == "abcdefghi"
 
 
+def test_set_enable_zoom():
+ context = SparkExpectationsContext(product_id="product1", spark=spark)
+ context.set_enable_zoom(True)
+ assert context._enable_zoom is True
+ assert context.get_enable_zoom is True
+
+
+def test_set_zoom_webhook_url():
+ context = SparkExpectationsContext(product_id="product1", spark=spark)
+ context.set_zoom_webhook_url("abcdefghi")
+ assert context._zoom_webhook_url == "abcdefghi"
+ assert context.get_zoom_webhook_url == "abcdefghi"
+
+
+def test_set_zoom_token():
+ context = SparkExpectationsContext(product_id="product1", spark=spark)
+ context.set_zoom_token("abcdefghi")
+ assert context._zoom_token == "abcdefghi"
+ assert context.get_zoom_token == "abcdefghi"
+
+
 def test_table_name():
  context = SparkExpectationsContext(product_id="product1", spark=spark)
  context.set_table_name("test_table")
@@ -709,6 +736,28 @@ def test_get_teams_webhook_url_exception():
  context.get_teams_webhook_url
 
 
+def test_get_zoom_webhook_url_exception():
+ context = SparkExpectationsContext(product_id="product1", spark=spark)
+ context._zoom_webhook_url = False
+ with pytest.raises(
+ SparkExpectationsMiscException,
+ match="The spark expectations context is not set completely, please assign "
+ "'_zoom_webhook_url' before \n accessing it",
+ ):
+ context.get_zoom_webhook_url
+
+
+def test_get_zoom_token():
+ context = SparkExpectationsContext(product_id="product1", spark=spark)
+ context._zoom_token = False
+ with pytest.raises(
+ SparkExpectationsMiscException,
+ match="The spark expectations context is not set completely, please assign "
+ "'_zoom_token' before \n accessing it",
+ ):
+ context.get_zoom_token
+
+
 def test_get_table_name_expection():
  context = SparkExpectationsContext(product_id="product1", spark=spark)
  context._table_name = ""