Notify Provider

The notify provider defines how workflow/task notifications are emitted.

Available providers

• NotifyDefault: no external notification channel.
• NotifyTelegram: sends notifications to Telegram.

Example

Python 3.8+

from dotflow import Config, DotFlow, action
from dotflow.core.types.status import TypeStatus
from dotflow.providers import NotifyDefault, NotifyTelegram


@action
def task():
    return {"notify": True}


def main():
    # No external notifications.
    workflow_default_notify = DotFlow(config=Config(notify=NotifyDefault()))
    workflow_default_notify.task.add(step=task)

    # Telegram notifications only for failed tasks.
    workflow_telegram_notify = DotFlow(
        config=Config(
            notify=NotifyTelegram(
                token="YOUR_BOT_TOKEN",
                chat_id=123456789,
                notification_type=TypeStatus.FAILED,
            )
        )
    )
    workflow_telegram_notify.task.add(step=task)

    return workflow_default_notify, workflow_telegram_notify


if __name__ == "__main__":
    main()

👀 Full file preview

Python 3.8+

from dotflow import Config, DotFlow, action
from dotflow.core.types.status import TypeStatus
from dotflow.providers import NotifyDefault, NotifyTelegram


@action
def task():
    return {"notify": True}


def main():
    # No external notifications.
    workflow_default_notify = DotFlow(config=Config(notify=NotifyDefault()))
    workflow_default_notify.task.add(step=task)

    # Telegram notifications only for failed tasks.
    workflow_telegram_notify = DotFlow(
        config=Config(
            notify=NotifyTelegram(
                token="YOUR_BOT_TOKEN",
                chat_id=123456789,
                notification_type=TypeStatus.FAILED,
            )
        )
    )
    workflow_telegram_notify.task.add(step=task)

    return workflow_default_notify, workflow_telegram_notify


if __name__ == "__main__":
    main()

Telegram setup example:

Python 3.8+

import os
import time

from dotenv import load_dotenv

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


@action
def simple_step(initial_context):
    time.sleep(0.5)

    return initial_context.storage


@action
def simple_step_raise():
    raise RuntimeError("Fail!")


def main():
    load_dotenv()

    notify = NotifyTelegram(
        token=os.getenv("BOT_TOKEN", ""),
        chat_id=int(os.getenv("CHAT_ID", "0")),
        notification_type=TypeStatus.FAILED,
    )

    workflow = DotFlow(config=Config(notify=notify))
    workflow.task.add(step=simple_step, initial_context={"foo": "bar"})
    workflow.task.add(step=simple_step_raise)
    workflow.start()

    return workflow


if __name__ == "__main__":
    main()

👀 Full file preview

Python 3.8+

import os
import time

from dotenv import load_dotenv

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


@action
def simple_step(initial_context):
    time.sleep(0.5)

    return initial_context.storage


@action
def simple_step_raise():
    raise RuntimeError("Fail!")


def main():
    load_dotenv()

    notify = NotifyTelegram(
        token=os.getenv("BOT_TOKEN", ""),
        chat_id=int(os.getenv("CHAT_ID", "0")),
        notification_type=TypeStatus.FAILED,
    )

    workflow = DotFlow(config=Config(notify=notify))
    workflow.task.add(step=simple_step, initial_context={"foo": "bar"})
    workflow.task.add(step=simple_step_raise)
    workflow.start()

    return workflow


if __name__ == "__main__":
    main()

References

• NotifyDefault
• NotifyTelegram