Apache Airflow Telegram Provider

Common errors

• Whenever the telegram operator is running, telegram will not receive any messages. Check the log in airflow and you can see that the task is stuck at Using connection ID 'telegram_conn_id' for task execution. for more than ten minutes without throwing any errors.

  cause This often occurs on macOS or Docker deployments due to Python's `_scproxy._get_proxies()` being invoked, leading to tasks getting stuck indefinitely due to network proxy resolution issues.
  fix For Docker or Linux environments, try setting the environment variable `no_proxy='*'` for your Airflow components (scheduler, workers, webserver). For macOS, this is a known `cpython` issue; configuring `no_proxy='*'` might also help.

• ModuleNotFoundError: No module named 'airflow.providers.telegram'

  cause The `apache-airflow-providers-telegram` package has not been correctly installed or is not accessible to your Airflow environment components (scheduler, worker, webserver).
  fix Ensure you have installed the provider using `pip install apache-airflow-providers-telegram`. If running Airflow in a Dockerized environment, make sure to add this `pip install` command to your Dockerfile and rebuild your Airflow images. Confirm all Airflow components have the provider installed.

• airflow.exceptions.AirflowException: The telegram_conn_id is not specified. Please set the 'telegram_conn_id' parameter in the operator or configure a default 'telegram_default' connection.

  cause The `TelegramOperator` requires a Telegram connection ID to authenticate. By default, it looks for a connection named `telegram_default`, or you can specify a different `telegram_conn_id` parameter. This error indicates neither was found or correctly configured.
  fix Create a new connection in the Airflow UI (Admin -> Connections) with a 'Conn Id' of `telegram_default` (or your chosen ID) and 'Conn Type' set to 'Telegram'. Set your Telegram Bot Token in the 'Password' field. Alternatively, pass the `token` parameter directly to the `TelegramOperator` (though Airflow connections are generally preferred for security).

Warnings

• breaking Minimum Apache Airflow version requirement has increased. Provider version 4.x.x (including 4.9.4) requires Apache Airflow >= 2.11.0.
  Fix: Ensure your Apache Airflow installation is version 2.11.0 or higher. If upgrading from an older Airflow version, follow the official Airflow upgrade guide and run `airflow upgrade db` if prompted.
• breaking The underlying `python-telegram-bot` library was upgraded to version 20.0.0+ (currently >=20.2), which introduced backward-incompatible changes. Specifically, the `get_conn()` method in `TelegramHook` became a coroutine function. This affects users with custom hooks extending `TelegramHook`.
  Fix: If you have custom hooks extending `TelegramHook`, you must update them to handle `get_conn()` as an asynchronous function. Refer to the `python-telegram-bot` transition guide for details on adapting to version 20.x changes.
• deprecated Older provider versions had a breaking change related to the `apply_default` decorator removal, requiring Airflow 2.1.0+. While this is covered by the current minimum Airflow version, users on very old Airflow installations (pre-2.1.0) upgrading directly to modern provider versions might encounter issues.
  Fix: Upgrade Airflow to at least 2.1.0 before installing or upgrading to affected provider versions. The current provider (4.9.4) requires Airflow 2.11.0+, which inherently resolves this specific warning.

Install

• pip install apache-airflow-providers-telegram Install latest version

Imports

• TelegramOperator

  from airflow.providers.telegram.operators.telegram import TelegramOperator

• TelegramFileOperator

  from airflow.providers.telegram.operators.telegram import TelegramFileOperator

Quickstart

This quickstart demonstrates how to set up an Airflow DAG to send a message to Telegram using the `TelegramOperator`. You'll need to first create a Telegram bot via BotFather to get an API token, and then configure an Airflow connection named `telegram_default` with this token. The `chat_id` should be the ID of the chat, group, or channel where the message will be sent.

import os
from datetime import datetime

from airflow import DAG
from airflow.providers.telegram.operators.telegram import TelegramOperator

# Configure Telegram Connection in Airflow UI:
# Conn Id: telegram_default
# Conn Type: Telegram
# Password: <YOUR_TELEGRAM_BOT_TOKEN> (from BotFather)
# Host (optional): https://api.telegram.org (default)

# Replace with your actual chat ID or fetch dynamically
TELEGRAM_CHAT_ID = os.environ.get('TELEGRAM_CHAT_ID', '-1234567890') # Example: Group/Channel Chat ID, starts with '-' or User Chat ID
TELEGRAM_BOT_TOKEN = os.environ.get('TELEGRAM_BOT_TOKEN', 'YOUR_BOT_TOKEN') # Only if not using Airflow Connection

with DAG(
    dag_id='telegram_notification_dag',
    start_date=datetime(2023, 1, 1),
    schedule_interval=None,
    catchup=False,
    tags=['telegram', 'notification'],
) as dag:
    send_message_task = TelegramOperator(
        task_id='send_test_message',
        telegram_conn_id='telegram_default', # This uses the connection configured in Airflow UI
        chat_id=TELEGRAM_CHAT_ID,
        text='Hello from Airflow! This is a test message from {{ dag.dag_id }}.',
        token=TELEGRAM_BOT_TOKEN if TELEGRAM_BOT_TOKEN != 'YOUR_BOT_TOKEN' else None # Optional: override connection token
    )