amorfo77/parsedmarc
1
0
Fork 0
mirror of https://github.com/domainaware/parsedmarc.git synced 2026-02-17 07:03:58 +00:00
Code Issues Projects Releases Wiki Activity

Enhance type hints and argument formatting in webhook.py for improved clarity and consistency

Browse Source
This commit is contained in:
Sean Whalen
2025-12-02 15:52:31 -05:00
parent d017dfcddf
commit 1127f65fbb
1 changed files with 21 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
parsedmarc/webhook.py
View File

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