Last active
August 25, 2021 20:19
-
-
Save fabidick22/db69dc868ae173bbf850131c026e865a to your computer and use it in GitHub Desktop.
Send notification to Slack from a AWS Lambda function with Webhook (no python dependencies).
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from datetime import datetime | |
from urllib.error import URLError, HTTPError | |
from urllib.parse import urlencode | |
from urllib.request import Request, urlopen | |
import json | |
import os | |
import logging | |
logger = logging.getLogger() | |
logger.setLevel(logging.INFO) | |
WEBHOOK_URL = os.environ['kmsEncryptedHookUrl'] | |
SLACK_CHANNEL = os.environ['slackChannel'] | |
HEADERS_FOR_REQUEST = ('Content-Type', 'application/json') | |
notification_types = { | |
"success": {"name": "Success", "color": "#36a64f"}, | |
"warning": {"name": "Warning", "color": "#F3C500"}, | |
"info": {"name": "Info", "color": "#b3b3b3"}, | |
"danger": {"name": "Danger", "color": "#F35A00"} | |
} | |
def lambda_handler(event, context): | |
logger.info("Event: " + str(event)) | |
logger.info("Context: " + str(context)) | |
message = event if type(event) is dict else json.loads(event) | |
logger.info("Message: " + str(message)) | |
sm = SimpleMessage("Alert", json.dumps(message).encode('utf-8'), "args.pretext", "success") | |
ns = NotificationSlack(sm, WEBHOOK_URL) | |
ns.send_message() | |
class NotificationSlack: | |
""" | |
Class for send notifications to a channel of Slack. | |
""" | |
def __init__(self, msg, url_app): | |
self.message = msg | |
self.url_app = url_app if url_app is not None else WEBHOOK_URL | |
def send_message(self): | |
""" | |
Function for send a message to channel. | |
:return (bool): Return True or False if the message was sent. | |
""" | |
data_to_send = json.dumps(self.message.get_template()) | |
data_to_send = data_to_send.replace("\\\\", "\\") | |
if isinstance(self.message, SimpleMessage) or isinstance(self.message, SSimpleMessage): | |
return True if self.make_request(data_to_send, self.url_app) else False | |
else: | |
raise Exception("Message type not supported.") | |
@staticmethod | |
def make_request(data, url): | |
""" | |
Function for creating request POST for sending a message | |
:param (str) data: Dict like json to be sent. | |
:param (str) url: Webhook URLs for Your Workspace. | |
:return (object): Return an object of a request. | |
""" | |
req = Request(url, data.encode("utf-8")) | |
req.add_header(*HEADERS_FOR_REQUEST) | |
try: | |
response = urlopen(req) | |
response.read() | |
logger.info("Message posted to %s", SLACK_CHANNEL) | |
except HTTPError as e: | |
logger.error("Request failed: %d %s", e.code, e.reason) | |
except URLError as e: | |
logger.error("Server connection failed: %s", e.reason) | |
return response | |
class CustomizingMessage: | |
pass | |
class Message: | |
""" | |
Super class Message | |
""" | |
def __init__(self, title, content, type_notification="info"): | |
self.title = title | |
self.content = content | |
self.type_notification = type_notification | |
@staticmethod | |
def check_type(data): | |
""" | |
Function for check type of message to build. | |
:param (str) data: String for type of message | |
:return (str): Return string of type message | |
:raise: Exception of type message no found. | |
""" | |
lower_data = data.lower() | |
if lower_data in notification_types: | |
return lower_data | |
else: | |
raise Exception("Type of message no found.") | |
class SSimpleMessage(Message): | |
""" | |
Class for build a very simple message for send like a message to Slack. | |
""" | |
def __init__(self, content, type_notification="info", title=""): | |
super().__init__(title=title, content=content, type_notification=type_notification) | |
def get_template(self): | |
""" | |
Function for getting a template like an object. | |
:return (dict): Dict of the template. | |
""" | |
return { | |
"attachments": [ | |
{ | |
"title": "{}".format(self.title), | |
"text": "{}".format(self.content), | |
"color": "{}".format(notification_types.get(self.type_notification)["color"]) | |
} | |
] | |
} | |
class SimpleMessage(Message): | |
""" | |
Class for build a simple message for send like a message to Slack. | |
""" | |
def __init__(self, title, content="", pre_title="", type_notification="info", date=datetime.today()): | |
super().__init__(title=title, content=content, type_notification=type_notification) | |
self.pre_title = pre_title | |
self.date = date | |
def get_template(self): | |
""" | |
Function for getting a template like an object. | |
:return (dict): Dict of the template. | |
""" | |
return { | |
"attachments": [ | |
{ | |
"title": "{}".format(self.title), | |
"pretext": "{}".format(self.pre_title), | |
"text": "{}".format(self.content), | |
"fields": [ | |
{ | |
"title": "Type", | |
"value": "{}".format(notification_types.get(self.type_notification)["name"]), | |
"short": True | |
}, | |
{ | |
"title": "Date", | |
"value": "{}".format(self.date.strftime("%Y-%m-%d %H:%M:%S")), | |
"short": True | |
} | |
], | |
"color": "{}".format(notification_types.get(self.type_notification)["color"]) | |
} | |
] | |
} | |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Notification:
