diff --git a/parsedmarc/webhook.py b/parsedmarc/webhook.py index 205407d..982808a 100644 --- a/parsedmarc/webhook.py +++ b/parsedmarc/webhook.py @@ -1,3 +1,11 @@ +# -*- coding: utf-8 -*- + +from __future__ import annotations + +from typing import Any, Optional, Union + +from collections import OrderedDict + import requests from parsedmarc import logger @@ -7,7 +15,13 @@ from parsedmarc.constants import USER_AGENT class WebhookClient(object): """A client for webhooks""" - def __init__(self, aggregate_url, forensic_url, smtp_tls_url, timeout=60): + def __init__( + self, + aggregate_url: str, + forensic_url: str, + smtp_tls_url: str, + timeout: Optional[int] = 60, + ): """ Initializes the WebhookClient Args: @@ -26,25 +40,27 @@ class WebhookClient(object): "Content-Type": "application/json", } - def save_forensic_report_to_webhook(self, report): + def save_forensic_report_to_webhook(self, report: OrderedDict[str, Any]): try: self._send_to_webhook(self.forensic_url, report) except Exception as error_: logger.error("Webhook Error: {0}".format(error_.__str__())) - def save_smtp_tls_report_to_webhook(self, report): + def save_smtp_tls_report_to_webhook(self, report: OrderedDict[str, Any]): try: self._send_to_webhook(self.smtp_tls_url, report) except Exception as error_: logger.error("Webhook Error: {0}".format(error_.__str__())) - def save_aggregate_report_to_webhook(self, report): + def save_aggregate_report_to_webhook(self, report: OrderedDict[str, Any]): try: self._send_to_webhook(self.aggregate_url, report) except Exception as error_: logger.error("Webhook Error: {0}".format(error_.__str__())) - def _send_to_webhook(self, webhook_url, payload): + def _send_to_webhook( + self, webhook_url: str, payload: Union[bytes, str, dict[str, Any]] + ): try: self.session.post(webhook_url, data=payload, timeout=self.timeout) except Exception as error_: