Como otimizar o desempenho de um aplicativo Flask

Introdução

Flask é uma estrutura web leve e flexível para a construção de aplicativos de pequeno e médio porte. É comumente usado em projetos que vão desde simples blogs pessoais até aplicativos mais complexos, como APIs REST, plataformas SaaS, sites de comércio eletrônico e painéis baseados em dados.

No entanto, à medida que o tráfego do seu aplicativo aumenta ou aumenta a complexidade, você pode começar a notar gargalos de desempenho. Esteja você construindo um sistema de gerenciamento de conteúdo (CMS), uma API para um aplicativo móvel ou uma ferramenta de visualização de dados em tempo real, otimizar o desempenho do Flask torna-se crucial para fornecer uma experiência de usuário responsiva e escalonável.

Neste tutorial, você explorará várias técnicas e práticas recomendadas para otimizar o desempenho de um aplicativo Flask.

Pré-requisitos

• Um servidor rodando Ubuntu e um usuário não root com privilégios sudo e um firewall ativo. Para obter orientação sobre como configurar isso, escolha sua distribuição nesta lista e siga nosso guia de configuração inicial do servidor. Certifique-se de trabalhar com uma versão compatível do Ubuntu.

• Familiaridade com a linha de comando do Linux. Para uma introdução ou atualização sobre a linha de comando, você pode visitar este guia no Linux Command Line Primer

• Uma compreensão básica da programação Python.

• Python 3.7 ou superior instalado em seu sistema Ubuntu. Para aprender como executar um script Python no Ubuntu, você pode consultar nosso tutorial sobre Como executar um script Python no Ubuntu.

Configurando seu ambiente Flask

O Ubuntu 24.04 vem com o Python 3 por padrão. Abra o terminal e execute o seguinte comando para verificar novamente a instalação do Python 3:

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

Se o Python 3 já estiver instalado em sua máquina, o comando acima retornará a versão atual da instalação do Python 3. Caso não esteja instalado, você pode executar o seguinte comando e obter a instalação do Python 3:

root@ubuntu:~# sudo apt install python3

Em seguida, você precisa instalar o instalador do pacote pip em seu sistema:

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

Depois que o pip estiver instalado, vamos instalar o Flask.

Você instalará o Flask via pip. É recomendado fazer isso em um ambiente virtual para evitar conflitos com outros pacotes do seu sistema.

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

root@ubuntu:~# pip install Flask

Crie um aplicativo Flask

A próxima etapa é escrever o código Python para o aplicativo Flask. Para criar um novo script, navegue até o diretório de sua preferência:

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

Quando estiver dentro do diretório, crie um novo arquivo Python, app.py, e importe o Flask. Em seguida, inicialize um aplicativo Flask e crie uma rota básica.

root@ubuntu:~# nano app.py

Isso abrirá um editor de texto em branco. Escreva sua lógica aqui ou copie o seguinte código:

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": "user@example.com"}
    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)

Agora, vamos executar o aplicativo Flask:

root@ubuntu:~# flask run

Você pode testar os endpoints com os seguintes comandos curl:

Teste o endpoint / (veicula conteúdo estático):

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

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

Teste o endpoint /slow (simula uma resposta lenta):

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

Para verificar esse endpoint lento, usamos o comando time no Linux. O comando time é usado para medir o tempo de execução de um determinado comando ou programa. Ele fornece três informações principais:

1. Tempo real: o tempo real decorrido do início ao fim do comando.
2. Tempo do usuário: a quantidade de tempo de CPU gasto no modo de usuário.
3. Tempo do sistema: a quantidade de tempo de CPU gasto no modo kernel.

Isso nos ajudará a medir o tempo real gasto pelo nosso ponto final lento. A saída pode ser algo assim:

{"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 solicitação leva cerca de 2 segundos para responder devido à chamada time.sleep(2) simulando uma resposta lenta.

Vamos testar o endpoint /db (simula uma operação de banco de dados):

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

{"email":"user@example.com","name":"User"}

Ao testar esses endpoints usando curl, você pode verificar se seu aplicativo Flask está sendo executado corretamente e se as respostas são as esperadas.

Na próxima seção você aprenderá como otimizar o desempenho dos aplicativos usando diversas técnicas.

Use um servidor WSGI pronto para produção

O servidor de desenvolvimento integrado do Flask não foi projetado para ambientes de produção. Para lidar com solicitações simultâneas de forma eficiente, você deve mudar para um servidor WSGI pronto para produção como o Gunicorn.

Instalar e configurar o Gunicorn

Vamos instalar o Gunicorn

root@ubuntu:~# pip install gunicorn

Execute o aplicativo Flask usando Gunicorn com 4 processos de trabalho:

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

 % /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

Aqui estão os benefícios de usar Gunicorn:

• Tratamento de solicitações simultâneas: Gunicorn permite que múltiplas solicitações sejam processadas simultaneamente usando vários processos de trabalho.
• Balanceamento de carga: equilibra as solicitações recebidas nos processos de trabalho, garantindo a utilização ideal dos recursos do servidor.
• Trabalhadores Assíncronos: Com trabalhadores assíncronos como gevent, ele pode lidar com eficiência com tarefas de longa execução sem bloquear outras solicitações.
• Escalabilidade: o Gunicorn pode escalar horizontalmente, aumentando o número de processos de trabalho para lidar com mais solicitações simultâneas.
• Tolerância a falhas: substitui automaticamente trabalhadores que não respondem ou travaram, garantindo alta disponibilidade.
• Pronto para produção: Ao contrário do servidor de desenvolvimento Flask, o Gunicorn é otimizado para ambientes de produção com melhores recursos de segurança, estabilidade e desempenho.

Ao mudar para o Gunicorn para produção, você pode melhorar significativamente o rendimento e a capacidade de resposta do seu aplicativo Flask, deixando-o pronto para lidar com o tráfego do mundo real com eficiência.

Habilite o cache para reduzir a carga

O cache é uma das melhores maneiras de melhorar o desempenho do Flask, reduzindo o processamento redundante. Aqui, você adicionará Flask-Caching para armazenar em cache o resultado da rota /slow.

Instalar e configurar o Flask-Caching com Redis

Instale os pacotes necessários:

root@ubuntu:~# pip install Flask-Caching redis

Atualize app.py para adicionar cache à rota /slow

Abra o editor e atualize o arquivo app.py com o seguinte:

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!")

Após a primeira solicitação para /slow, as solicitações subsequentes dentro de 60 segundos serão atendidas a partir do cache, ignorando a função time.sleep(). Isso reduz a carga do servidor e acelera os tempos de resposta.

Para verificar se os dados estão sendo armazenados em cache, vamos executar os comandos abaixo para o endpoint /slow.

Esta é a primeira solicitação para o endpoint /slow. Após a conclusão desta solicitação, o resultado da rota /slow é armazenado em cache.

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

{"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 é uma solicitação subsequente para o endpoint /slow dentro de 60 segundos:

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

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

Otimize consultas de banco de dados

As consultas ao banco de dados muitas vezes podem se tornar um gargalo de desempenho. Nesta seção, você simulará a otimização de consultas de banco de dados usando SQLAlchemy e pool de conexões.

Simular uma consulta de banco de dados com pool de conexões

Primeiro, vamos instalar o SQLAlchemy

root@ubuntu:~# pip install Flask-SQLAlchemy

Atualize app.py para configurar o pool de conexões

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))

Agora, quando executamos uma solicitação curl para a rota db1, devemos observar a seguinte saída:

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

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

Você pode otimizar significativamente o desempenho do seu aplicativo Flask implementando o pool de conexões em um ambiente de produção. O pool de conexões permite que o aplicativo reutilize conexões de banco de dados existentes em vez de criar novas para cada solicitação. Isso reduz a sobrecarga de estabelecimento de novas conexões, levando a tempos de resposta mais rápidos e melhor escalabilidade.

A configuração SQLALCHEMY_POOL_SIZE que definimos anteriormente limita o número de conexões no pool. Você deve ajustar esse valor em um ambiente de produção com base nos requisitos específicos e nos recursos do servidor. Além disso, você pode considerar outras opções de pool, como SQLALCHEMY_MAX_OVERFLOW para permitir conexões extras quando o pool estiver cheio e SQLALCHEMY_POOL_TIMEOUT para definir quanto tempo uma solicitação aguardará por uma conexão.

Lembre-se, embora nosso exemplo use SQLite para simplificar, em um cenário do mundo real, você provavelmente usaria um banco de dados mais robusto como PostgreSQL ou MySQL. Esses bancos de dados têm seus próprios mecanismos de pooling de conexões que podem ser aproveitados em conjunto com o pooling do SQLAlchemy para um desempenho ainda melhor.

Ao configurar e utilizar cuidadosamente o pool de conexões, você pode garantir que seu aplicativo Flask lide com as operações de banco de dados com eficiência, mesmo sob alta carga, melhorando significativamente seu desempenho geral.

Habilitar compactação Gzip

A compactação de suas respostas pode reduzir drasticamente a quantidade de dados transferidos entre seu servidor e clientes, melhorando o desempenho.

Instalar e configurar o Flask-Compress

Vamos instalar o pacote Flask-compress.

root@ubuntu:~# pip install Flask-Compress

A seguir, vamos atualizar app.py para ativar a compactação.

from flask_compress import Compress
This below command enables Gzip compression for the Flask appIt 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>"

Isso compactará automaticamente respostas maiores que 500 bytes, reduzindo o tempo de transferência para respostas grandes.

Em um ambiente de produção, a compactação Gzip pode reduzir significativamente a quantidade de dados transferidos entre o servidor e os clientes, especialmente para conteúdo baseado em texto como HTML, CSS e JavaScript.

Essa redução na transferência de dados leva a tempos de carregamento de página mais rápidos, melhor experiência do usuário e custos reduzidos de largura de banda. Além disso, muitos navegadores modernos suportam automaticamente a descompressão Gzip, tornando-a uma técnica de otimização amplamente compatível. Ao habilitar a compactação Gzip, você pode melhorar efetivamente o desempenho e a escalabilidade do seu aplicativo Flask sem exigir nenhuma alteração do lado do cliente.

Transferir tarefas intensivas para o aipo

Para operações com muitos recursos, como envio de e-mails ou processamento de grandes conjuntos de dados, é melhor transferi-los para tarefas em segundo plano usando Celery. Isso evita que tarefas de longa duração bloqueiem solicitações recebidas.

O Celery é um poderoso sistema de fila de tarefas distribuídas que permite executar tarefas demoradas de forma assíncrona. Ao transferir operações intensivas para o Celery, você pode melhorar significativamente a capacidade de resposta e escalabilidade do seu aplicativo Flask. O Celery funciona delegando tarefas a processos de trabalho, que podem ser executados em máquinas separadas, permitindo melhor utilização de recursos e processamento paralelo.

Os principais benefícios do uso do aipo incluem:

1. Tempos de resposta aprimorados para solicitações de usuários
2. Melhor escalabilidade e gerenciamento de recursos
3. Capacidade de lidar com tarefas complexas e demoradas sem bloquear o aplicativo principal
4. Suporte integrado para agendamento de tarefas e repetição de tarefas com falha
5. Fácil integração com vários corretores de mensagens como RabbitMQ ou Redis

Ao aproveitar o Celery, você pode garantir que seu aplicativo Flask permaneça responsivo mesmo ao lidar com tarefas computacionalmente intensivas ou vinculadas a E/S.

Configurar o aipo para tarefas em segundo plano

Vamos instalar o Celery.

root@ubuntu:~# pip install Celery

A seguir, vamos atualizar app.py para configurar o Celery para tarefas assíncronas:

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'

Em um terminal separado, inicie o trabalhador Celery:

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

------------- 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.

Agora execute um comando curl para acessar a rota /start-task, a saída será:

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

Task started

Isso retornaria “Tarefa iniciada ” quase instantaneamente, mesmo que a tarefa em segundo plano ainda estivesse em execução.

A função start_task() faz duas coisas:

• Ele chama long_task.delay(), que inicia de forma assíncrona a tarefa do Celery. Isso significa que a tarefa está na fila para ser executada em segundo plano, mas a função não espera que ela seja concluída.

• Ele retorna imediatamente a string ‘Tarefa iniciada’.

O importante a notar é que a tarefa real de longa duração (simulada pelo sono de 10 segundos) é executada de forma assíncrona pelo Celery. A rota Flask não espera a conclusão desta tarefa antes de responder à solicitação.

Portanto, ao enrolar esse endpoint, você receberá uma resposta imediata dizendo “Tarefa iniciada”, enquanto a tarefa real continua sendo executada em segundo plano por 10 segundos.

Após 10 segundos, quando a tarefa em segundo plano for concluída, você deverá notar esta mensagem de log:

A saída será semelhante a esta:

[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 exemplo mostra como o Celery melhora o desempenho do aplicativo Flask lidando com tarefas de longa execução de forma assíncrona, mantendo a capacidade de resposta do aplicativo principal. A longa tarefa será executada em segundo plano, liberando o aplicativo Flask para lidar com outras solicitações.

Em um ambiente de produção, a implementação do Celery envolve:

1. Usando um corretor de mensagens robusto como RabbitMQ
2. Empregando um back-end de resultados dedicado (por exemplo, PostgreSQL)
3. Gerenciando trabalhadores com sistemas de controle de processo (por exemplo, Supervisor)
4. Implementando ferramentas de monitoramento (por exemplo, Flower)
5. Aprimorando o tratamento e registro de erros
6. Utilizando a priorização de tarefas
7. Dimensionamento com vários trabalhadores em máquinas diferentes
8. Garantindo medidas de segurança adequadas

Conclusão

Neste tutorial, você aprendeu como otimizar um aplicativo Flask implementando várias técnicas de melhoria de desempenho. Seguindo essas etapas, você pode melhorar o desempenho, a escalabilidade e a capacidade de resposta do seu aplicativo Flask, garantindo que ele seja executado com eficiência mesmo sob carga pesada.