backup-tool/app/notifier.py

79 lines
2.2 KiB
Python
Raw Normal View History

2021-04-26 17:45:19 +02:00
"""Module handling IFTTT notifications."""
2021-01-15 21:29:23 +01:00
from typing import List
import os
import requests
2021-01-15 21:08:56 +01:00
class Notifier:
2021-04-26 17:45:19 +02:00
"""A notifier object that can send IFTTT notifications."""
2021-01-15 21:29:23 +01:00
# (positive, negative)
_EVENTS = {
"backup": (
"Backup for {name} succeeded.",
"Backup for {name} failed.",
),
"restore": (
"{name} successfully restored.",
"Couldn't restore {name}.",
),
}
2021-04-26 17:45:19 +02:00
"""The message content for a given event."""
2021-01-15 21:29:23 +01:00
2021-01-15 21:08:56 +01:00
# Placeholder
2021-01-15 21:29:23 +01:00
def __init__(
self, title: str, events: List[str], endpoint: str, api_key: str = None
):
2021-04-26 17:45:19 +02:00
"""Initialize a new Notifier object.
Args:
title: the notification title to use
events: the event types that should trigger a notification (should
be one of the keys in _EVENTS).
endpoint: IFTTT endpoint name
api_key: your IFTTT API key. If not provided, it will be read from
the IFTTT_API_KEY environment variable.
Todo:
* Read the API key on init
"""
2021-01-15 21:29:23 +01:00
self.title = title
self.events = events
self.endpoint = endpoint
self.api_key = api_key
def notify(self, category: str, name: str, status_code: int):
2021-04-26 17:45:19 +02:00
"""Send an IFTTT notification.
2021-01-15 21:29:23 +01:00
Args:
2021-04-26 17:45:19 +02:00
category: type of notify (should be one of the keys in _EVENTS).
Only if the category was passed during initialization will the
notification be sent.
2021-01-15 21:29:23 +01:00
name: name of the spec
status_code: exit code of the command
"""
event = "{}_{}".format(
category, "success" if status_code == 0 else "failure"
)
# stop if it's not a valid event
if event not in self._EVENTS:
return
api_key = self.api_key or os.environ["IFTTT_API_KEY"]
# Can't do anything without a key
if not api_key:
return
url = " https://maker.ifttt.com/trigger/{}/with/key/{}".format(
self.endpoint, api_key
)
data = {
"value1": self.title,
"value2": self._EVENTS[event][int(status_code != 0)],
}
requests.post(url, data=data)