Skip to content

Notifications on Failure

Configure your workflow to send notifications only when tasks fail. Useful for monitoring production pipelines without being flooded by success messages.

Example with Telegram

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


@action
def extract():
    return {"data": fetch_data()}


@action
def transform(previous_context):
    return {"result": process(previous_context.storage)}


@action
def load(previous_context):
    save(previous_context.storage)
    return {"loaded": True}


def main():
    config = Config(
        storage=StorageFile(),
        notify=NotifyTelegram(
            token="YOUR_BOT_TOKEN",
            chat_id=123456789,
            notification_type=TypeStatus.FAILED,
        ),
    )

    workflow = DotFlow(config=config)
    workflow.task.add(step=extract)
    workflow.task.add(step=transform)
    workflow.task.add(step=load)
    workflow.start()


if __name__ == "__main__":
    main()

How it works

  1. NotifyTelegram is configured with notification_type=TypeStatus.FAILED
  2. After each task status changes, dotflow calls notify.hook_status_task(task)
  3. The provider checks if task.status matches the configured notification_type
  4. Only failed tasks trigger a Telegram message

Notification types

Type When notified
TypeStatus.FAILED Only on failure
TypeStatus.COMPLETED Only on success
None (default) Every task, regardless of status

References