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.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

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.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

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

1
2
3
4
5
6
7

@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