diff --git a/alertify.py b/alertify.py index 6cb03b5..4506891 100644 --- a/alertify.py +++ b/alertify.py @@ -12,7 +12,7 @@ from src import Alertify if __name__ == '__main__': - def parse_cli(): + def parse_cli() -> argparse.ArgumentParser: """ Function to parse the CLI """ @@ -45,9 +45,9 @@ if __name__ == '__main__': return parser.parse_args() - def main(): + def main() -> int: """ - main() + Main program logic """ logging.basicConfig( format='%(levelname)s: %(message)s', @@ -69,7 +69,6 @@ if __name__ == '__main__': # ----------------------------- if args.healthcheck: - # Invert the sense of 'healthy' for Unix CLI usage _, status = alertify.healthcheck() return status == 200 diff --git a/src/Alertify/__init__.py b/src/Alertify/__init__.py index 66c8363..1404da3 100644 --- a/src/Alertify/__init__.py +++ b/src/Alertify/__init__.py @@ -11,6 +11,7 @@ __version__ = '2.0' import json import logging +from typing import Optional, Tuple import werkzeug.exceptions from flask import Flask, request, request_started @@ -39,7 +40,7 @@ class Alertify(FlaskView): # Instantiate with defaults self.configure() - def configure(self, configfile=None): + def configure(self, configfile: Optional[str] = None): """ Configure from a configfile """ @@ -71,7 +72,7 @@ class Alertify(FlaskView): webapp.run(host='0.0.0.0', port=self.config.listen_port) @route('/alert', methods=['POST']) - def alert(self): + def alert(self) -> Tuple[str, int]: """ Handle the alerts from Alertmanager """ @@ -89,7 +90,7 @@ class Alertify(FlaskView): except UnboundLocalError: return '', 204 - def healthcheck(self): + def healthcheck(self) -> Tuple[str, int]: """ Perform a healthcheck and return the results """ diff --git a/src/Alertify/config.py b/src/Alertify/config.py index e4fbf1c..e5269da 100644 --- a/src/Alertify/config.py +++ b/src/Alertify/config.py @@ -5,6 +5,7 @@ import inspect import logging import os from distutils.util import strtobool +from typing import Optional import yaml @@ -23,7 +24,7 @@ class Config: listen_port = int(8080) verbose = int(0) - def __init__(self, configfile=None): + def __init__(self, configfile: Optional[str] = None): """ Method to parse a configuration file """ @@ -50,14 +51,14 @@ class Config: else: setattr(self, key, type(default_val)(userval)) - def items(self): + def items(self) -> list: """ Method to return an iterator for the configured items """ return {key: getattr(self, key) for key in self.__dict__}.items() @classmethod - def keys(cls): + def keys(cls) -> list: """ Method to return the defaults as a list of dict_keys """ @@ -74,7 +75,7 @@ class Config: ] @classmethod - def defaults(cls): + def defaults(cls) -> dict: """ Classmethod to return the defaults as a dictionary """ diff --git a/src/Alertify/gotify.py b/src/Alertify/gotify.py index 794de71..f1aac5b 100644 --- a/src/Alertify/gotify.py +++ b/src/Alertify/gotify.py @@ -5,6 +5,7 @@ import http.client import json import logging import socket +from typing import Optional class Gotify: @@ -12,7 +13,7 @@ class Gotify: Class to handle Gotify communications """ - def __init__(self, server, port, app_key, client_key=None): + def __init__(self, server: str, port: int, app_key: str, client_key: Optional[str] = None): self.api = http.client.HTTPConnection(server, port) self.app_key = app_key self.client_key = client_key @@ -21,7 +22,7 @@ class Gotify: 'Accept': 'application/json', } - def _call(self, method, url, body=None): + def _call(self, method: str, url: str, body: Optional[str] = None) -> dict: """ Method to call Gotify with an app or client key as appropriate """ @@ -60,14 +61,14 @@ class Gotify: return resp_obj - def delete(self, msg_id): + def delete(self, msg_id: str) -> dict: """ Method to delete a message from the Gotify server """ logging.debug('Deleting message ID: %s', msg_id) return self._call('DELETE', f'/message/{msg_id}') - def find_byfingerprint(self, message): + def find_byfingerprint(self, message: str) -> list: """ Method to return the ID of a matching message """ @@ -75,7 +76,7 @@ class Gotify: new_fingerprint = message['fingerprint'] except KeyError: logging.debug('No fingerprint found in new message') - return None + return list() msg_list = [] for old_message in self.messages(): @@ -91,7 +92,7 @@ class Gotify: return msg_list - def messages(self): + def messages(self) -> dict: """ Method to return a list of messages from the Gotify server """ @@ -103,14 +104,14 @@ class Gotify: logging.debug('Fetching existing messages from Gotify') return self._call('GET', '/message')['json'].get('messages', []) - def send_alert(self, payload): + def send_alert(self, payload: dict) -> dict: """ Method to send a message payload to a Gotify server """ logging.debug('Sending message to Gotify') return self._call('POST', '/message', body=json.dumps(payload, indent=2)) - def healthcheck(self): + def healthcheck(self) -> dict: """ Method to perform a healthcheck against Gotify """ diff --git a/src/Alertify/health.py b/src/Alertify/health.py index f96abb1..40121fb 100644 --- a/src/Alertify/health.py +++ b/src/Alertify/health.py @@ -1,6 +1,9 @@ """ Module for handling any healthcheck related activity """ +from typing import Tuple + +from .gotify import Gotify class Healthcheck: @@ -8,10 +11,10 @@ class Healthcheck: Class to handle the healthchecks """ - def __init__(self, gotify_client): + def __init__(self, gotify_client: Gotify): self.gotify = gotify_client - def gotify_alive(self): + def gotify_alive(self) -> Tuple[str, int]: """ Simple method to return the Gotify healthcheck response """ diff --git a/src/Alertify/messaging.py b/src/Alertify/messaging.py index 993f39f..6a543ee 100644 --- a/src/Alertify/messaging.py +++ b/src/Alertify/messaging.py @@ -2,6 +2,9 @@ Module for handling the messaging """ import logging +from typing import Optional + +from .gotify import Gotify class MessageHandler: @@ -9,12 +12,17 @@ class MessageHandler: Class to handle alert messaging """ - def __init__(self, gotify_client, disable_resolved=False, delete_onresolve=False): + def __init__( + self, + gotify_client: Gotify, + disable_resolved: Optional[bool] = False, + delete_onresolve: Optional[bool] = False, + ): self.gotify = gotify_client self.disable_resolved = disable_resolved self.delete_onresolve = delete_onresolve - def process(self, alert): + def process(self, alert: dict) -> dict: """ Method to process the alert message """