Source code for herostools.actor.webhook

import json
from typing import Any

import requests
from jinja2 import Template

from .archiver.base import HERODataArchiver




class HEROWebhookSender(HERODataArchiver):
    """HERODataArchiver that renders metadata and payload into a JSON template
    and posts them to an arbitrary incoming-webhook endpoint.

    Works with any webhook-compatible platform (Slack, Mattermost, Discord,
    MS Teams, ...) since the payload format is fully defined by the template.

    Args:
        object_selector: Zenoh object selector for the devices to subscribe to.
        event_name: Name of the event.
        webhook_url: Full URL of the incoming webhook endpoint.
        default_metadata: Default metadata merged with every incoming payload's metadata.
        body_template: Template for POST message body as JSON string rendered with Jinja2,
            using the payload and metadata as template variables (e.g.
            ``'{"text": {{ payload.value | tojson }}}'``). Rendering payload only works, if
            it is json serializable.
        max_retries: Number of times to retry a failed delivery before dropping the item.
        request_timeout: Timeout in seconds for the POST request to the webhook endpoint.
    """

    def __init__(
        self,
        webhook_url: str,
        body_template: str,
        request_timeout: float = 10,
        *args,
        **kwargs,
    ):
        # compile the template up front to fail early on template syntax errors
        self._payload_in_template = "payload" in body_template
        self._webhook_url = webhook_url
        self._template = Template(body_template)
        self.request_timeout = request_timeout
        HERODataArchiver.__init__(self, *args, **kwargs)
        # If the template wants to insert the payload add a marker to insert later on



    def _store(self, source_name: str, payload: Any, metadata: dict) -> None:
        """Send the payload array as webhook.

        Args:
            source_name: Name of the event source (the HERO).
            payload: Incoming data.
            metadata: Incoming metadata merged with ``default_metadata``.
        """
        # Allow to send json serialized payload, if wanted.
        # May fail if payload is not serializable
        if self._payload_in_template:
            metadata["payload"] = payload if isinstance(payload, dict) else json.dumps(payload)
        message = self._template.render(metadata)
        body = json.loads(message)
        response = requests.post(self._webhook_url, json=body, timeout=self.request_timeout)
        response.raise_for_status()