SUEN

Handbook: Implementing Non-Blocking Telegram Notifications in a Synchronous Flask/Gunicorn Application

1. The Problem: App Crashes and Unexpected token '<' Errors

When attempting to add a feature like sending a Telegram notification after a web request, a common mistake is to introduce code that conflicts with the application’s execution model. In our case, we encountered an Unexpected token '<', "<!DOCTYPE "... is not valid JSON error on the frontend.

Diagnosis: This error indicates that the frontend expected a JSON response from the API but received an HTML error page instead. This happens when the backend Flask application crashes mid-request due to an unhandled exception.

The root cause of the crash was attempting to run asynchronous code (from the python-telegram-bot library using asyncio) inside a synchronous Flask application managed by Gunicorn’s synchronous workers. This creates a conflict between event loops and process models, leading to instability and crashes.

2. The Wrong Solutions & Why They Failed

3. The Correct, Robust Solution: Threading and a Synchronous HTTP Client

The goal is to send the notification without blocking the main application thread from returning a response to the user. The most reliable way to achieve this in a standard Flask/Gunicorn setup is by offloading the task to a separate background thread.

The principles are:

  1. Keep the Main App Synchronous: The core Flask application should remain simple and synchronous.
  2. Use a Synchronous Library for the Task: Instead of a complex async library, use the standard requests library to make a simple, synchronous HTTP call to the Telegram Bot API.
  3. Offload to a Background Thread: Use Python’s built-in threading module to run the notification function in the background. This immediately frees up the main thread to finish the request.

4. Implementation Steps

Step 1: Use a Simple, Synchronous Function for Notifications

Create a standard Python function that uses requests to call the Telegram API. This function is self-contained and easy to debug.

python 
import requests
import logging
import os

TELEGRAM_BOT_TOKEN = os.getenv("TELEGRAM_BOT_TOKEN")
TELEGRAM_CHAT_ID = os.getenv("TELEGRAM_CHAT_ID")

def send_telegram_notification(message: str):
    """
    Sends a message to a Telegram chat using the synchronous requests library.
    """
    if not TELEGRAM_BOT_TOKEN or not TELEGRAM_CHAT_ID:
        logging.warning("Telegram not configured, skipping notification.")
        return

    api_url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage"
    payload = {
        "chat_id": TELEGRAM_CHAT_ID,
        "text": message,
        "parse_mode": "MarkdownV2"
    }
    
    try:
        # Use a short timeout to prevent the thread from hanging indefinitely.
        response = requests.post(api_url, json=payload, timeout=10)
        response.raise_for_status() # Raise an exception for HTTP error codes
        logging.info("Successfully sent notification to Telegram.")
    except requests.RequestException as e:
        logging.error(f"Failed to send notification to Telegram: {e}")

Step 2: Use a Decorator to Wrap the Flask Route Handler

A decorator is an elegant way to add logging and the notification logic without modifying the core business logic of the route handler itself.

python 
import functools
import threading
from flask import request

def log_and_notify(f):
    @functools.wraps(f)
    def decorated_function(*args, **kwargs):
        start_time = time.time()
        response = None
        status_code = 500
        
        try:
            # Execute the original route handler (e.g., your login logic)
            response = f(*args, **kwargs)
            status_code = response[1] if isinstance(response, tuple) else response.status_code
            return response
        except Exception:
            # Handle crashes within the route
            # ... (return a 500 error response)
            pass
        finally:
            # This block always runs, after the response has been determined.
            
            # ... (prepare your audit_info dictionary here) ...
            
            # --- Key Step: Offload the notification to a background thread ---
            if TELEGRAM_BOT_TOKEN and TELEGRAM_CHAT_ID:
                # Format the message for Telegram
                tg_message = format_my_telegram_message(audit_info, response)

                # Create and start the thread
                tg_thread = threading.Thread(
                    target=send_telegram_notification, 
                    args=(tg_message,)
                )
                tg_thread.daemon = True # Allows the main app to exit without waiting for the thread
                tg_thread.start()
                
    return decorated_function

Step 3: Apply the Decorator to Your Route

python 
@app.route('/attendance', methods=['POST'])
@log_and_notify
def handler():
    # Your core business logic remains clean and untouched here.
    # It just returns a Flask response.
    # ...
    return jsonify({"status": "SUCCESS", "message": "Task complete."}), 200

5. Key Takeaways to Avoid Future Errors