Cómo optimizar el rendimiento de una aplicación Flask

Introducción

Flask es un framework web ligero y flexible para crear aplicaciones pequeñas y medianas. Se utiliza comúnmente en proyectos que abarcan desde blogs personales sencillos hasta aplicaciones más complejas, como API REST, plataformas SaaS, sitios web de comercio electrónico y paneles de control basados ​​en datos.

Sin embargo, a medida que aumenta el tráfico o la complejidad de su aplicación, podría notar cuellos de botella en el rendimiento. Ya sea que esté desarrollando un sistema de gestión de contenido (CMS), una API para una aplicación móvil o una herramienta de visualización de datos en tiempo real, optimizar el rendimiento de Flask es crucial para ofrecer una experiencia de usuario ágil y escalable.

En este tutorial, explorará técnicas y mejores prácticas para optimizar el rendimiento de una aplicación Flask.

Requisitos previos

• Un servidor con Ubuntu y un usuario no root con privilegios de sudo y un firewall habilitado. Para obtener instrucciones sobre cómo configurarlo, seleccione su distribución de esta lista y siga nuestra guía "Introducción a un servidor". Asegúrese de utilizar una versión compatible de Ubuntu.
• Introducción a la línea de comandos de Linux Para una introducción o repaso de la línea de comandos, puede consultar esta guía sobre Introducción a la línea de comandos de Linux.
• Comprensión básica de la programación en Python.
• Tiene instalado Python 3.7 o superior en su sistema Ubuntu. Para aprender a ejecutar scripts de Python en Ubuntu, puede consultar nuestro tutorial sobre cómo ejecutar scripts de Python en Ubuntu.

Configuración de su entorno Flask

Ubuntu 24.04 viene con Python 3 por defecto. Abra una terminal y ejecute el siguiente comando para comprobar su instalación de Python 3:

root@ubuntu:~# python3 --version
Python 3.12.3

Si Python 3 ya está instalado en su equipo, el comando anterior devolverá la versión actual de la instalación de Python 3. Si no está instalado, puede ejecutar el siguiente comando para obtener la instalación de Python 3:

root@ubuntu:~# sudo apt install python3

A continuación, debe instalar el instalador del paquete pip en su sistema:

root@ubuntu:~# sudo apt install python3-pip

Después de instalar pip, instalemos Flask.

Instalará Flask mediante pip. Se recomienda hacerlo en un entorno virtual para evitar interferencias con otros paquetes del sistema.

root@ubuntu:~# python3 -m venv myprojectenv
root@ubuntu:~# source myprojectenv/bin/activate

root@ubuntu:~# pip install Flask

Crear una aplicación Flask

El siguiente paso es escribir el código Python para la aplicación Flask. Para crear un nuevo script, vaya al directorio que desee:

root@ubuntu:~# cd ~/path-to-your-script-directory

Una vez en el directorio, crea un nuevo archivo de Python, app.py, e importa Flask. A continuación, inicializa una aplicación Flask y crea una ruta raíz.

root@ubuntu:~# nano app.py

Se abrirá un editor de texto en blanco. Escribe tu lógica aquí o copia el código a continuación:

from flask import Flask, jsonify, request
app = Flask(__name__)
# Simulate a slow endpoint
@app.route('/slow')
def slow():
import time
time.sleep(2) # to simulate a slow response
return jsonify(message="This request was slow!")
# Simulate an intensive database operation
@app.route('/db')
def db_operation():
# This is a dummy function to simulate a database query
result = {"name": "User", "email": "[email protected]"}
return jsonify(result)
# Simulate a static file being served
@app.route('/')
def index():
return "<h1>Welcome to the Sample Flask App</h1>"
if __name__ == '__main__':
app.run(debug=True)

Ahora ejecutemos la aplicación Flask:

root@ubuntu:~# flask run

Puede probar los puntos finales con los siguientes comandos curl:

Pruebe el punto final / (devuelve contenido estático):

root@ubuntu:~# curl http://127.0.0.1:5000/

[secondary_lebel Output]
<h1>Welcome to the Sample Flask App</h1>%

Prueba el punto final/lento (simula una respuesta lenta):

root@ubuntu:~# time curl http://127.0.0.1:5000/db

Para comprobar este punto final lento, utilizamos el comando time en Linux. Este comando mide el tiempo de ejecución de un comando o programa específico. Proporciona tres datos principales:

1. Tiempo real: El tiempo real transcurrido desde el inicio hasta el final del comando.
2. Tiempo de usuario: cantidad de tiempo de CPU empleado en modo usuario.
3. Tiempo del sistema: la cantidad de tiempo de CPU empleado en modo kernel.

Esto nos ayudará a medir el tiempo real que tarda nuestro punto final lento. El resultado podría ser similar a esto:

Output
{"message":"This request was slow!"}
curl http://127.0.0.1:5000/slow 0.00s user 0.01s system 0% cpu 2.023 total

Esta solicitud tarda aproximadamente 2 segundos en responder, porque la llamada time.sleep(2) simula una respuesta lenta.

Probemos el punto final /db (simula operaciones de base de datos):

root@ubuntu:~# curl http://127.0.0.1:5000/db

Output
{"email":"[email protected]","name":"User"}

Al probar estos puntos finales con curl, puede verificar que su aplicación Flask se esté ejecutando correctamente y que las respuestas sean las esperadas.

En la siguiente sección, aprenderá a optimizar el rendimiento de la aplicación utilizando diversas técnicas.

Utilice un servidor WSGI listo para producción

El servidor de desarrollo Flask integrado no está diseñado para entornos de producción. Para gestionar eficazmente las solicitudes concurrentes, conviene migrar a un servidor WSGI listo para producción, como Gunicorn.

Instalar y ejecutar Gunicorn

Vamos a instalar Gunicorn.

root@ubuntu:~# pip install gunicorn

Ejecute la aplicación Flask usando Gunicorn con 4 procesos de trabajo:

root@ubuntu:~# gunicorn -w 4 -b 0.0.0.0:8000 app:app

Output
% /Library/Python/3.9/bin/gunicorn -w 4 -b 0.0.0.0:8000 app:app
[2024-09-13 18:37:24 +0530] [99925] [INFO] Starting gunicorn 23.0.0
[2024-09-13 18:37:24 +0530] [99925] [INFO] Listening at: http://0.0.0.0:8000 (99925)
[2024-09-13 18:37:24 +0530] [99925] [INFO] Using worker: sync
[2024-09-13 18:37:24 +0530] [99926] [INFO] Booting worker with pid: 99926
[2024-09-13 18:37:25 +0530] [99927] [INFO] Booting worker with pid: 99927
[2024-09-13 18:37:25 +0530] [99928] [INFO] Booting worker with pid: 99928
[2024-09-13 18:37:25 +0530] [99929] [INFO] Booting worker with pid: 99929
[2024-09-13 18:37:37 +0530] [99925] [INFO] Handling signal: winch
^C[2024-09-13 18:38:51 +0530] [99925] [INFO] Handling signal: int
[2024-09-13 18:38:51 +0530] [99927] [INFO] Worker exiting (pid: 99927)
[2024-09-13 18:38:51 +0530] [99926] [INFO] Worker exiting (pid: 99926)
[2024-09-13 18:38:51 +0530] [99928] [INFO] Worker exiting (pid: 99928)
[2024-09-13 18:38:51 +0530] [99929] [INFO] Worker exiting (pid: 99929)
[2024-09-13 18:38:51 +0530] [99925] [INFO] Shutting down: Master

Estos son los beneficios de usar Gunicorn:

• Manejo de solicitudes simultáneas: Gunicorn permite procesar múltiples solicitudes simultáneamente utilizando múltiples procesos de trabajo.
• Equilibrio de carga: equilibra las solicitudes entrantes entre los procesos de trabajo y garantiza un uso óptimo de los recursos del servidor.
• Trabajadores asincrónicos: con trabajadores asincrónicos como gevent, se pueden ejecutar de manera eficiente tareas de larga ejecución sin bloquear otras solicitudes.
• Escalabilidad: Gunicorn puede escalar horizontalmente aumentando la cantidad de procesos de trabajo para manejar más solicitudes simultáneas.
• Tolerancia a fallos: reemplaza automáticamente a los trabajadores que no responden o que están dañados, lo que garantiza una alta disponibilidad.
• Listo para producción: a diferencia del servidor de desarrollo Flask, Gunicorn está optimizado para entornos de producción con mejores características de seguridad, estabilidad y rendimiento.

Al cambiar a Gunicorn para producción, puede mejorar significativamente el rendimiento y la capacidad de respuesta de su aplicación Flask, preparándola para manejar de manera eficiente el tráfico del mundo real.

Habilite el almacenamiento en caché para reducir la carga.

El almacenamiento en caché es una de las mejores maneras de mejorar el rendimiento de Flask al reducir la sobrecarga. Aquí, se agrega Flask-Caching para almacenar en caché el resultado de la ruta lenta.

Instalar y configurar Flask-Caching con Redis

Instalar los paquetes necesarios:

Actualizar app.py para agregar caché a slow/path

Abra el editor y actualice el archivo app.py con el siguiente comando:

root@ubuntu:~# nano app.py

from flask_caching import Cache
app = Flask(__name__)
# Configure Flask-Caching with Redis
app.config['CACHE_TYPE'] = 'redis'
app.config['CACHE_REDIS_HOST'] = 'localhost'
app.config['CACHE_REDIS_PORT'] = 6379
cache = Cache(app)
@app.route('/slow')
@cache.cached(timeout=60)
def slow():
import time
time.sleep(2) # Simulate a slow response
return jsonify(message="This request was slow!")

Tras la primera solicitud a /slow, las solicitudes posteriores se procesan desde la caché en un plazo de 60 segundos, sin pasar por la función time.sleep(). Esto reduce la carga del servidor y aumenta la velocidad de respuesta.

Nota: En este tutorial, utilizamos localhost como host de Redis. Sin embargo, en un entorno de producción, se recomienda usar un servicio de Redis administrado como DigitalOcean Managed Redis. Esto proporciona mayor escalabilidad, confiabilidad y seguridad para sus necesidades de almacenamiento. Puede obtener más información sobre la integración de DigitalOcean Managed Redis en una aplicación de producción en este tutorial sobre almacenamiento con DigitalOcean Redis en la plataforma de aplicaciones.

Para comprobar si los datos se están almacenando en caché, ejecutamos los siguientes comandos para el punto final /slow.

Esta es la primera solicitud al punto final /slow. Una vez completada, el resultado de la ruta /slow se almacena en caché.

root@ubuntu:~# time curl http://127.0.0.1:5000/slow

Output
{"message":"This request was slow!"}
curl http://127.0.0.1:5000/slow 0.00s user 0.01s system 0% cpu 2.023 total

Esta es una solicitud posterior al punto final /slow dentro de los 60 segundos:

root@ubuntu:~# time curl http://127.0.0.1:5000/slow

Output
{"message":"This request was slow!"}
curl http://127.0.0.1:5000/slow 0.00s user 0.00s system 0% cpu 0.015 total

Optimizar las consultas de la base de datos

Las consultas a bases de datos a menudo pueden convertirse en un cuello de botella en el rendimiento. En esta sección, simulará la optimización de consultas a bases de datos mediante SQLAlchemy y la agrupación de uniones.

Simulación de una consulta de base de datos con agregación de unión

Primero, instalemos SQLAlchemy.

root@ubuntu:~# pip install Flask-SQLAlchemy

Actualice app.py para configurar la integración de la conexión.

rom flask_sqlalchemy import SQLAlchemy
from sqlalchemy import text
# Simulate an intensive database operation
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///test.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.config['SQLALCHEMY_POOL_SIZE'] = 5 # Connection pool size
db = SQLAlchemy(app)
@app.route('/db1')
def db_operation_pooling():
# Simulate a database query
result = db.session.execute(text('SELECT 1')).fetchall()
return jsonify(result=str(result))

Ahora, cuando ejecutamos la solicitud curl a la ruta db1, deberíamos observar el siguiente resultado:

root@ubuntu:~# curl http://127.0.0.1:5000/db1 

output
{"result":"[(1,)]"}

Puede optimizar significativamente el rendimiento de su aplicación Flask implementando la agrupación de conexiones en un entorno de producción. Esta agrupación permite que la aplicación reutilice las conexiones de base de datos existentes en lugar de crear nuevas conexiones para cada solicitud. Esto reduce la sobrecarga de crear nuevas conexiones, lo que se traduce en tiempos de respuesta más rápidos y una mejor escalabilidad.

La configuración SQLALCHEMY_POOL_SIZE que establecimos anteriormente limita el número de conexiones en el pool. Debe ajustar este valor en un entorno de producción según sus necesidades específicas y las capacidades del servidor. Además, puede considerar otras opciones de integración como SQLALCHEMY_MAX_OVERFLOW para permitir conexiones adicionales si el pool está lleno y SQLALCHEMY_POOL_TIMEOUT para especificar cuánto tiempo debe esperar una solicitud para obtener una conexión.

Recuerde que, si bien nuestro ejemplo usa SQLite para simplificar, en un escenario real, probablemente esté usando una base de datos más robusta como PostgreSQL o MySQL. Estas bases de datos cuentan con sus propios mecanismos de agrupación de conexiones que pueden usarse junto con la integración de SQLAlchemy para un mejor rendimiento.

Con una configuración cuidadosa y el uso del agrupamiento de conexiones, puede garantizar que su aplicación Flask realice operaciones de base de datos de manera eficiente incluso bajo una carga pesada, mejorando así significativamente su rendimiento general.

Habilitar la compresión Gzip

Comprimir sus respuestas puede reducir drásticamente la cantidad de datos transferidos entre su servidor y sus clientes y mejorar el rendimiento.

Instalar y configurar Flask-Compress

Instalemos el paquete Flask-compress.

root@ubuntu:~# pip install Flask-Compress

A continuación, actualicemos app.py para habilitar la compresión.

from flask_compress import Compress
# This below command enables Gzip compression for the Flask app
# It compresses responses before sending them to clients,
# reducing data transfer and improving performance
Compress(app)
@app.route('/compress')
def Compress():
return "<h1>Welcome to the optimized Flask app !</h1>"

Esto comprime automáticamente las respuestas de más de 500 bytes, lo que reduce el tiempo de transmisión de respuestas grandes.

En un entorno de producción, la compresión Gzip puede reducir significativamente la cantidad de datos transferidos entre el servidor y los clientes, especialmente para contenido basado en texto como HTML, CSS y JavaScript.

Esta reducción en la transferencia de datos se traduce en tiempos de carga de páginas más rápidos, una mejor experiencia de usuario y menores costos de ancho de banda. Además, muchos navegadores web modernos admiten automáticamente la compresión Gzip, lo que la convierte en una técnica de optimización totalmente compatible. Al habilitar la compresión Gzip, puede mejorar eficazmente el rendimiento y la escalabilidad de su aplicación Flask sin tener que realizar cambios en el cliente.

Subir tareas intensivas a Celery

Para operaciones que consumen muchos recursos, como el envío de correos electrónicos o el procesamiento de grandes conjuntos de datos, es mejor transferirlas a trabajos en segundo plano con Celery. Esto evita que las tareas de larga duración bloqueen las solicitudes entrantes.

Celery es un potente sistema de colas de tareas distribuidas que permite ejecutar tareas que consumen mucho tiempo de forma asíncrona. Al delegar operaciones intensivas a Celery, se puede mejorar significativamente la capacidad de respuesta y la escalabilidad de la aplicación Flask. Celery funciona delegando tareas a procesos de trabajo que pueden ejecutarse en máquinas independientes, lo que permite un mejor uso de los recursos y el procesamiento en paralelo.

Los principales beneficios del uso del apio incluyen:

1. Mejorar el tiempo de respuesta a las solicitudes de los usuarios
2. Mejor escalabilidad y gestión de recursos
3. Capacidad de realizar tareas complejas y que requieren mucho tiempo sin bloquear el programa principal
4. Soporte integrado para la programación de tareas y reintento de tareas fallidas
5. Fácil integración con varios agentes de mensajes como RabbitMQ o Redis

Al usar Celery, puede garantizar que su aplicación Flask siga respondiendo incluso cuando se trata de tareas que requieren un uso intensivo de recursos informáticos o están limitadas por E/S.

Configurar Celery para tareas en segundo plano

Vamos a instalar Celery.

root@ubuntu:~# pip install Celery

A continuación, actualicemos app.py para configurar Celery para tareas asincrónicas:

from celery import Celery
celery = Celery(app.name, broker='redis://localhost:6379/0')
@celery.task
def long_task():
import time
time.sleep(10) # Simulate a long task
return "Task Complete"
@app.route('/start-task')
def start_task():
long_task.delay()
return 'Task started'

En una terminal separada, inicie el trabajador Celery:

root@ubuntu:~# celery -A app.celery worker --loglevel=info

Output
------------- celery@your-computer-name v5.2.7 (dawn-chorus)
--- ***** ----- 
-- ******* ---- Linux-x.x.x-x-generic-x86_64-with-glibc2.xx 2023-xx-xx
- *** --- * --- 
- ** ---------- [config]
- ** ---------- .> app: app:0x7f8b8c0b3cd0
- ** ---------- .> transport: redis://localhost:6379/0
- ** ---------- .> results: disabled://
- *** --- * --- .> concurrency: 8 (prefork)
-- ******* ---- .> task events: OFF (enable -E to monitor tasks in this worker)
--- ***** ----- 
-------------- [queues]
.> celery exchange=celery(direct) key=celery
[tasks]
. app.long_task
[2023-xx-xx xx:xx:xx,xxx: INFO/MainProcess] Connected to redis://localhost:6379/0
[2023-xx-xx xx:xx:xx,xxx: INFO/MainProcess] mingle: searching for neighbors
[2023-xx-xx xx:xx:xx,xxx: INFO/MainProcess] mingle: all alone
[2023-xx-xx xx:xx:xx,xxx: INFO/MainProcess] celery@your-computer-name ready.

Ahora ejecute un comando curl para acceder a la ruta /start-task, el resultado será el siguiente:

root@ubuntu:~# curl http://127.0.0.1:5000/start-task

Output
Task started

Esto devolverá "Tarea iniciada" casi inmediatamente, incluso si el trabajo en segundo plano todavía se está ejecutando.

La función start_task() hace dos cosas:

• Llama a long_task.delay(), que inicia la tarea de Celery de forma asíncrona. Esto significa que la tarea se pone en cola para ejecutarse en segundo plano, pero la función no espera a que se complete.
• Devuelve inmediatamente la cadena "Tarea iniciada".

Es importante tener en cuenta que la tarea de larga duración (simulada por la suspensión de 10 segundos) se ejecuta asincrónicamente en Celery. La ruta Flask no espera a que esta tarea se complete para responder a la solicitud.

Entonces, cuando se atornilla a este punto final, se obtiene inmediatamente una respuesta que dice "trabajo iniciado", mientras que el trabajo real continúa ejecutándose en segundo plano durante 10 segundos.

Después de 10 segundos, cuando se complete el trabajo en segundo plano, debería ver este mensaje de informe:

La salida será similar a esto:

[2024-xx-xx xx:xx:xx,xxx: INFO/MainProcess] Task app.long_task[task-id] received
[2024-xx-xx xx:xx:xx,xxx: INFO/ForkPoolWorker-1] Task app.long_task[task-id] succeeded in 10.xxxs: 'Task Complete'

Este ejemplo muestra cómo Celery mejora el rendimiento de una aplicación Flask al ejecutar tareas de larga duración de forma asíncrona, manteniendo la aplicación principal activa. La tarea de larga duración se ejecuta en segundo plano, liberando así la aplicación Flask para gestionar otras solicitudes.

En un entorno de producción, ejecutar Celery implica lo siguiente:

1. Usando un agente de mensajes robusto como RabbitMQ
2. Uso de un backend de resultados propietario (por ejemplo, PostgreSQL)
3. Gestionar a los trabajadores con sistemas de control de procesos (por ejemplo, supervisor)
4. Implementar herramientas de monitoreo (por ejemplo, flores) Aumentar el manejo y registro de errores
5. Mayor gestión y registro de errores
6. Utilice la priorización de tareas
7. Escalado con múltiples trabajadores en diferentes máquinas
8. Garantizar medidas de seguridad adecuadas

Resultado

En este tutorial, aprendiste a optimizar tu aplicación Flask mediante diversas técnicas de mejora del rendimiento. Siguiendo estos pasos, puedes mejorar el rendimiento, la escalabilidad y la capacidad de respuesta de tu aplicación Flask, garantizando su eficiente funcionamiento incluso con cargas pesadas.