Skip to content

NotifyDiscord

dotflow.providers.notify_discord.NotifyDiscord

Bases: Notify

Import

You can import the NotifyDiscord class with:

from dotflow.providers import NotifyDiscord
Example

class dotflow.providers.notify_discord.NotifyDiscord

from dotflow import Config, DotFlow
from dotflow.providers import NotifyDiscord
from dotflow.core.types.status import TypeStatus

config = Config(
    notify=NotifyDiscord(
        webhook_url="https://discord.com/api/webhooks/...",
        notification_type=TypeStatus.FAILED,
    )
)

workflow = DotFlow(config=config)

Parameters:

Name Type Description Default
webhook_url str

Discord webhook URL.

 required
notification_type Optional[TypeStatus]

Filter notifications by task status. If None, all statuses are notified.

 None
show_result bool

Include task result in the notification. Defaults to False.

 False
timeout float

Request timeout in seconds.

 1.5
Source code in dotflow/providers/notify_discord.py 
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
class NotifyDiscord(Notify):
    """
    Import:
        You can import the **NotifyDiscord** class with:

            from dotflow.providers import NotifyDiscord

    Example:
        `class` dotflow.providers.notify_discord.NotifyDiscord

            from dotflow import Config, DotFlow
            from dotflow.providers import NotifyDiscord
            from dotflow.core.types.status import TypeStatus

            config = Config(
                notify=NotifyDiscord(
                    webhook_url="https://discord.com/api/webhooks/...",
                    notification_type=TypeStatus.FAILED,
                )
            )

            workflow = DotFlow(config=config)

    Args:
        webhook_url (str): Discord webhook URL.

        notification_type (Optional[TypeStatus]): Filter notifications
            by task status. If None, all statuses are notified.

        show_result (bool): Include task result in the notification.
            Defaults to False.

        timeout (float): Request timeout in seconds.
    """

    COLORS = {
        TypeStatus.COMPLETED: 0x4CAF50,
        TypeStatus.FAILED: 0xF44336,
        TypeStatus.RETRY: 0xFF9800,
        TypeStatus.IN_PROGRESS: 0x2196F3,
        TypeStatus.NOT_STARTED: 0x9E9E9E,
        TypeStatus.PAUSED: 0x607D8B,
    }

    def __init__(
        self,
        webhook_url: str,
        notification_type: TypeStatus | None = None,
        show_result: bool = False,
        timeout: float = 1.5,
    ):
        self.webhook_url = webhook_url
        self.notification_type = notification_type
        self.show_result = show_result
        self.timeout = timeout

    def hook_status_task(self, task: Any) -> None:
        if self.notification_type and self.notification_type != task.status:
            return

        try:
            response = post(
                url=self.webhook_url,
                headers={"Content-Type": "application/json"},
                data=dumps({"embeds": [self._build_embed(task)]}),
                timeout=self.timeout,
            )
            response.raise_for_status()
        except Exception as error:
            logger.error(
                "Internal problem sending notification on Discord: %s",
                str(error),
            )

    def _build_embed(self, task: Any) -> dict:
        embed = {
            "title": f"{TypeStatus.get_symbol(task.status)} {task.status}",
            "color": self.COLORS.get(task.status, 0x9E9E9E),
            "description": f"`{task.workflow_id}` — Task {task.task_id}",
            "fields": [],
        }

        if self.show_result:
            embed["fields"].append(
                {
                    "name": "Result",
                    "value": f"```json\n{task.result(max=1024)}```",
                }
            )

        if task.status == TypeStatus.FAILED and task.errors:
            last_error = task.errors[-1]
            error_value = f"`{last_error.exception}`: {last_error.message}"
            embed["fields"].append(
                {"name": "Error", "value": error_value[:1024]}
            )

        return embed

COLORS = {TypeStatus.COMPLETED: 5025616, TypeStatus.FAILED: 16007990, TypeStatus.RETRY: 16750592, TypeStatus.IN_PROGRESS: 2201331, TypeStatus.NOT_STARTED: 10395294, TypeStatus.PAUSED: 6323595} class-attribute instance-attribute

notification_type = notification_type instance-attribute

show_result = show_result instance-attribute

timeout = timeout instance-attribute

webhook_url = webhook_url instance-attribute

__init__(webhook_url, notification_type=None, show_result=False, timeout=1.5)

Source code in dotflow/providers/notify_discord.py 
59
60
61
62
63
64
65
66
67
68
69
def __init__(
    self,
    webhook_url: str,
    notification_type: TypeStatus | None = None,
    show_result: bool = False,
    timeout: float = 1.5,
):
    self.webhook_url = webhook_url
    self.notification_type = notification_type
    self.show_result = show_result
    self.timeout = timeout

hook_status_task(task)

Source code in dotflow/providers/notify_discord.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
def hook_status_task(self, task: Any) -> None:
    if self.notification_type and self.notification_type != task.status:
        return

    try:
        response = post(
            url=self.webhook_url,
            headers={"Content-Type": "application/json"},
            data=dumps({"embeds": [self._build_embed(task)]}),
            timeout=self.timeout,
        )
        response.raise_for_status()
    except Exception as error:
        logger.error(
            "Internal problem sending notification on Discord: %s",
            str(error),
        )