Add type hints

This commit is contained in:
Scott Wallace 2021-05-25 10:06:09 +01:00
parent 1e0099400d
commit 4d74be6b9f
6 changed files with 36 additions and 23 deletions

View file

@ -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

View file

@ -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
"""

View file

@ -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
"""

View file

@ -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
"""

View file

@ -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
"""

View file

@ -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
"""