Cómo construir y desplegar un bot de python-telegram v20 con Webhook

La versión v20 de python-telegram-bot introdujo grandes cambios estructurales. Según la documentación, toda la lógica relacionada con la red y la E/S ahora funciona a través de funciones de corutina (es decir, funciones async def ...). En particular, todos los métodos de la clase telegram.Bot que realizan solicitudes a la API del bot ahora son funciones de corutina y necesitan.

La versión 20 de python-telegram-bot introdujo cambios estructurales importantes. Según la documentación,

“Toda la lógica relacionada con la red y la E/S ahora funciona a través de funciones de coroutine (es decir, funciones async def ...). En particular, todos los métodos de la clase telegram.Bot que realizan solicitudes a la API del bot son ahora funciones de coroutine y deben ser esperados con await.” — historial de cambios de python-telegram-bot v20

El proceso de transición de la versión 13 de python-telegram-bot a la versión 20 resultó ser más complejo de lo que había anticipado inicialmente. Convertir funciones síncronas def a async def y añadir await a nuevas coroutines fue bastante fácil. Pero la dificultad principal fue encontrar una documentación detallada sobre cómo crear e implementar webhooks de python-telegram-bot v20 en un entorno de producción.

Este artículo explica por qué y cómo hice la migración de:

python-telegram-bot v13 → v20
Flask → FastAPI
Gunicorn → Gunicorn + Uvicorn

¿Por qué actualizar de v13 a v20?

Las versiones v13.x y anteriores ya no son compatibles con el equipo de desarrollo de python-telegram-bot. Si la API de Telegram introduce nuevas funciones, solo estarán disponible en la versión v20 o superior.

¿Por qué usar un Webhook y no la obtención de resultados mediante consultas (polling)?

La mayoría de los ejemplos proporcionados por el equipo de desarrollo de python-telegram-bot utilizan Application.run_polling. Pero generalmente se recomiendan los webhooks en lugar de la obtención de resultados mediante consultas (polling) en la mayoría de los casos de uso de los bots de Telegram, porque la obtención de resultados mediante consultas requiere que tu bot realice constantemente solicitudes a los servidores de Telegram, lo cual puede consumir muchos recursos. En cambio, los webhooks ofrecen funcionalidad extendida, actualizaciones más rápidas y mejor escalabilidad.

El desafío de usar Flask con python-telegram-bot v20

Usar un WSGI como Flask con python-telegram-bot v20 es incómodo.

Flask, una interfaz de paso en el servidor web (Web Server Gateway Interface, WSGI), es síncrona y solo puede manejar una solicitud a la vez. Pero aún puedes ejecutar funciones asíncronas en Flask utilizando asyncio.run(), como en el ejemplo de bot de webhook personalizado proporcionado por el equipo de desarrollo de python-telegram-bot.

asyncio.run() inicia un ciclo de eventos y ejecuta la coroutine dada hasta que se complete. Si hay tareas asíncronas que se ejecutan antes o después de que se maneje la solicitud, esas tareas se ejecutarán en un ciclo de eventos separado.

# Fragmento de código de https://docs.python-telegram-bot.org/en/v20.6/examples.customwebhookbot.htmlwebserver = uvicorn.Server(    config=uvicorn.Config(        app=WsgiToAsgi(flask_app),        port=PORT,        use_colors=False,        host="127.0.0.1",    ))async with application:  await application.start()  await webserver.serve() # start bot's webserver  await application.stop()

Sin embargo, esta implementación es un poco incómoda porque Flask es intrínsecamente incompatible con los globales asíncronos.

Los ejemplos en la documentación no son adecuados para entornos de producción.

Usar asyncio.run() como punto de entrada en producción generalmente no es recomendado. La función asyncio.run() está diseñada para propósitos de desarrollo y pruebas, y es posible que no proporcione el mismo nivel de robustez y confiabilidad que los servidores de producción como Gunicorn o UWSGI.

Estos servidores de producción ofrecen muchas características adicionales, como el registro, monitoreo y comprobaciones de salud, que son esenciales para garantizar la estabilidad y seguridad de una aplicación en producción.

Si deseas implementar tu bot en producción, es mucho más limpio utilizar ASGI (Interfaz de puerta de enlace de servidor asincrónico) con una implementación de servidor web ASGI.

Cómo hacerlo: migración y implementación

De Flask (WSGI) a FastAPI (AGSI)

Migrar una aplicación Flask a una ASGI es sencillo. Elegí FastAPI porque encontré un completo tutorial de migración aquí. La sintaxis de ambos frameworks es bastante similar, lo que significa que no tendrás que hacer demasiados cambios en el código.

# Desde python-telegram-bot v20application = (    Application.builder()    .updater(None)    .token(<tu-token-del-bot>) # reemplaza <tu-token-del-bot>    .read_timeout(7)    .get_updates_read_timeout(42)    .build())# Desde FastAPI@asynccontextmanagerasync def lifespan(app: FastAPI):    async with application:        await application.start()        yield        await application.stop()

Quart parece ser una alternativa factible, pero no ofrece soporte para implementaciones de Uvicorn, que es la implementación de servidor web que estaba adaptando del script proporcionado por el equipo de python-telegram-bot.

Un ejemplo funcional

El siguiente código muestra un ejemplo mínimo que utiliza FastAPI para construir un webhook de python-telegram-bot v20. Este bot responderá con “¡empezando…” cuando reciba el comando /start.

# main.pyfrom contextlib import asynccontextmanagerfrom http import HTTPStatusfrom telegram import Updatefrom telegram.ext import Application, CommandHandlerfrom telegram.ext._contexttypes import ContextTypesfrom fastapi import FastAPI, Request, Response# Inicializar el bot de Telegram en Pythonptb = (    Application.builder()    .updater(None)    .token(<tu-token-del-bot>) # reemplaza <tu-token-del-bot>    .read_timeout(7)    .get_updates_read_timeout(42)    .build())@asynccontextmanagerasync def lifespan(_: FastAPI):    await ptb.bot.setWebhook(<tu-url-del-webhook>) # reemplaza <tu-url-del-webhook>    async with ptb:        await ptb.start()        yield        await ptb.stop()# Inicializar la aplicación FastAPI (similar a Flask)app = FastAPI(lifespan=lifespan)@app.post("/")async def process_update(request: Request):    req = await request.json()    update = Update.de_json(req, ptb.bot)    await ptb.process_update(update)    return Response(status_code=HTTPStatus.OK)# Ejemplo de controladorasync def start(update, _: ContextTypes.DEFAULT_TYPE):    """Enviar un mensaje cuando se emite el comando /start."""    await update.message.reply_text("¡empezando...")ptb.add_handler(CommandHandler("start", start))

Para iniciar el bot, instala todas las dependencias necesarias usando pip y ejecuta el comando de inicio: gunicorn main:app -k uvicorn.workers.UvicornWorker.

Este fragmento de código está adaptado de un bot de Telegram real en producción. Consulta el código fuente de @cron_telebot aquí para ver cómo se implementa. Siéntete libre de adaptar el script según tu caso de uso.