Notify with Telegram

Bot Telegram

Create a bot on Telegram:

1 - Open the Telegram app on your mobile device.

2 - Tap the search icon and look for BotFather.

3 - Open the contact and type /newbot.

4 - Enter a name for your bot.

Generate Bot Token:

1 - Type the command /token.

2 - Select the bot you just created from the list.

3 - BotFather will return an access token — copy and save this token securely. It will be required to authenticate your bot.

Retrieving Your Telegram Chat ID

1 - Send a message to your bot in Telegram to ensure it appears in the bot's update log.

2 - Run the following curl command to fetch the latest updates from your bot:

curl --location --globoff 'https://api.telegram.org/bot<YOUR_BOT_TOKEN>/getUpdates'

Replace <YOUR_BOT_TOKEN> with the token provided by BotFather.

3 - Inspect the API response, and locate the following key result[0].channel_post.chat.id. This is your chat ID.

4 - Copy and store the chat ID in a secure location. You will need it to configure the NotifyTelegram instance.

DotFlow Config

Use the runnable example below:

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()