securehst_webhook_notifier package

securehst_webhook_notifier.notify_webhook(webhook_url: str, func_identifier: str, platform: str = 'mattermost', user_id: str | None = None, custom_message: str | None = None) → Callable[[Callable[[...], Any]], Callable[[...], Any]]

Decorator that sends start, end, and error notifications via a webhook.

Args:

webhook_url (str): The webhook URL to send notifications to. func_identifier (str): A string identifier representing the function being decorated. platform (str, optional): Messaging platform type: “mattermost”, “slack”, or “discord”. Defaults to “mattermost”. user_id (Optional[str], optional): User ID or username to mention on errors. Platform-specific formatting is applied. Defaults to None. custom_message (Optional[str], optional): Optional custom message to include. Defaults to None.

Returns:

Callable: A wrapped function with webhook notifications.

securehst_webhook_notifier.prefect_notify_webhook(webhook_url: str, display_name: str, user_id: str | None = None, silent_success: bool = True, platform: str = 'mattermost', start_message: str | None = None, success_message: str | None = None, failure_message: str | None = None) → Callable[[Callable[[...], Any]], Callable[[...], Any]]

Decorator to add comprehensive Prefect flow notifications.

This decorator should be applied BEFORE the @flow decorator to ensure proper integration with Prefect’s state management system.

Args:

webhook_url (str): The webhook URL to send notifications to. display_name (str): Human readable name for the flow (e.g., “D. Miller & Associates - dmiller-etl”). user_id (Optional[str], optional): User to mention on failures (e.g., “securehst”). Defaults to None. silent_success (bool, optional): If True, success notifications won’t mention users. Defaults to True. platform (str, optional): Messaging platform type: “mattermost”, “slack”, or “discord”.

Defaults to “mattermost”.

start_message (Optional[str], optional): Custom start message.

Defaults to detailed message with start time.

success_message (Optional[str], optional): Custom success message.

Defaults to detailed message with timing info.

failure_message (Optional[str], optional): Custom failure message.

Defaults to detailed message with timing and error info.

Returns:

Callable: A wrapped function with Prefect webhook notifications.

Raises:

ImportError: If Prefect is not available.

Example:

@prefect_notify_webhook(

webhook_url=”https://mattermost.example.com/hooks/abc123”, display_name=”ETL Pipeline”, user_id=”admin”

) @flow def my_etl_flow():

pass

securehst_webhook_notifier.send_webhook_message(webhook_url: str, message: str, platform: str) → None

Sends a formatted message to the specified webhook URL.

Args:

webhook_url (str): The destination webhook URL. message (str): The message content to send. platform (str): Platform type to determine payload structure (“mattermost”, “slack”, “discord”).

Raises:

ValueError: If the platform is unsupported. requests.RequestException: If the HTTP request fails.

Submodules

securehst_webhook_notifier.main module

securehst_webhook_notifier.main.create_state_hooks(webhook_url: str, display_name: str, user_id: str | None, silent_success: bool, start_time_holder: dict, platform: str = 'mattermost', success_message: str | None = None, failure_message: str | None = None) → dict

Create state change hooks for Prefect flow lifecycle notifications.

securehst_webhook_notifier.main.notify_webhook(webhook_url: str, func_identifier: str, platform: str = 'mattermost', user_id: str | None = None, custom_message: str | None = None) → Callable[[Callable[[...], Any]], Callable[[...], Any]]

Decorator that sends start, end, and error notifications via a webhook.

Args:

webhook_url (str): The webhook URL to send notifications to. func_identifier (str): A string identifier representing the function being decorated. platform (str, optional): Messaging platform type: “mattermost”, “slack”, or “discord”. Defaults to “mattermost”. user_id (Optional[str], optional): User ID or username to mention on errors. Platform-specific formatting is applied. Defaults to None. custom_message (Optional[str], optional): Optional custom message to include. Defaults to None.

Returns:

Callable: A wrapped function with webhook notifications.

securehst_webhook_notifier.main.prefect_notify_webhook(webhook_url: str, display_name: str, user_id: str | None = None, silent_success: bool = True, platform: str = 'mattermost', start_message: str | None = None, success_message: str | None = None, failure_message: str | None = None) → Callable[[Callable[[...], Any]], Callable[[...], Any]]

Decorator to add comprehensive Prefect flow notifications.

This decorator should be applied BEFORE the @flow decorator to ensure proper integration with Prefect’s state management system.

Args:

webhook_url (str): The webhook URL to send notifications to. display_name (str): Human readable name for the flow (e.g., “D. Miller & Associates - dmiller-etl”). user_id (Optional[str], optional): User to mention on failures (e.g., “securehst”). Defaults to None. silent_success (bool, optional): If True, success notifications won’t mention users. Defaults to True. platform (str, optional): Messaging platform type: “mattermost”, “slack”, or “discord”.

Defaults to “mattermost”.

start_message (Optional[str], optional): Custom start message.

Defaults to detailed message with start time.

success_message (Optional[str], optional): Custom success message.

Defaults to detailed message with timing info.

failure_message (Optional[str], optional): Custom failure message.

Defaults to detailed message with timing and error info.

Returns:

Callable: A wrapped function with Prefect webhook notifications.

Raises:

ImportError: If Prefect is not available.

Example:

@prefect_notify_webhook(

webhook_url=”https://mattermost.example.com/hooks/abc123”, display_name=”ETL Pipeline”, user_id=”admin”

) @flow def my_etl_flow():

pass

securehst_webhook_notifier.main.send_prefect_notification(webhook_url: str, message: str, platform: str = 'mattermost') → None

Send a notification to the webhook URL with error swallowing.

Args:

webhook_url (str): The destination webhook URL. message (str): The message content to send. platform (str): Platform type for payload structure (“mattermost”, “slack”, “discord”).

securehst_webhook_notifier.main.send_start_notification(webhook_url: str, display_name: str, start_time: datetime, platform: str = 'mattermost', start_message: str | None = None) → None

Send flow start notification with timestamp.

securehst_webhook_notifier.main.send_webhook_message(webhook_url: str, message: str, platform: str) → None

Sends a formatted message to the specified webhook URL.

Args:

webhook_url (str): The destination webhook URL. message (str): The message content to send. platform (str): Platform type to determine payload structure (“mattermost”, “slack”, “discord”).

Raises:

ValueError: If the platform is unsupported. requests.RequestException: If the HTTP request fails.