Neste artigo, vou guiar o desenvolvimento de uma aplicação web baseada em Flask que interage com um banco de dados SQL Server para analisar dados populacionais. A aplicação permite aos usuários consultar faixas de população, buscar condados por estado e recuperar estados dentro de faixas específicas de população. Também irei discutir como integrar o Redis para armazenar em cache os resultados das consultas e melhorar o desempenho.
Por que Flask, SQL Server e Redis?
O Flask é um framework web Python leve e flexível, perfeito para construir aplicações web de pequeno a médio porte. Ele fornece as ferramentas necessárias para criar APIs RESTful, renderizar templates HTML dinâmicos e interagir com bancos de dados. Por outro lado, o SQL Server é um robusto sistema de gerenciamento de banco de dados relacional (RDBMS) amplamente utilizado em aplicações corporativas. Combinar o Flask com o SQL Server nos permite construir uma aplicação poderosa para análise e visualização de dados.
Para melhorar ainda mais o desempenho, vamos integrar o Redis, um armazenamento de dados em memória, para armazenar em cache os resultados das consultas frequentemente acessadas. Isso reduz a carga no banco de dados e acelera os tempos de resposta para consultas repetidas.
Visão Geral da Aplicação
Nossa aplicação Flask realiza as seguintes tarefas:
- Intervalos de população de consulta. Os usuários podem especificar um ano e intervalo de população para obter contagens de estados que se enquadram dentro desses intervalos.
- Buscar condados por estado. Os usuários podem inserir um código de estado para recuperar uma lista de condados.
- Recuperar estados por intervalo de população. Os usuários podem especificar um intervalo de população e ano para obter uma lista de estados dentro desse intervalo.
- Nota. Para testar, fique à vontade para criar seu próprio esquema no banco de dados e inserir dados de exemplo conforme necessário com base nos seguintes APIs compartilhados usando consultas SQL. Além disso, as páginas HTML usadas aqui podem ser um design básico de tabela que captura os dados retornados do código do aplicativo Flask e exibe os resultados.
Vamos mergulhar nos detalhes da implementação.
Configurando a Aplicação Flask
1. Pré-requisitos
Antes de começar, certifique-se de ter o seguinte instalado através do seu terminal root (comandos compatíveis com o MacOS):
- Python 3.x
- Flask (
pip install flask) - SQLAlchemy (
pip install sqlalchemy) - PyODBC (
pip install pyodbc) - Redis (
pip install redis)
2. Conexão com o Banco de Dados
Nós usamos SQLAlchemy para conectar ao banco de dados SQL Server. Veja como a conexão pode ser configurada:
from sqlalchemy import create_engine
import urllib
# SQL Server connection string
params = urllib.parse.quote_plus(
"Driver={ODBC Driver 17 for SQL Server};"
"Server=tcp:username.database.windows.net,1433;"
"Database=population;"
"Uid=user@username;"
"Pwd={azure@123};"
"Encrypt=yes;"
"TrustServerCertificate=no;"
"Connection Timeout=30;"
)
engine = create_engine("mssql+pyodbc:///?odbc_connect=%s" % params)
Esta string de conexão utiliza o Driver ODBC para SQL Server e inclui parâmetros para criptografia e tempo limite.
3. Configuração do Redis
O Redis é utilizado para armazenar em cache os resultados das consultas. Veja como configurar a conexão com o Redis:
import redis
# Redis connection
redis_client = redis.StrictRedis(
host='username.redis.cache.windows.net',
port=6380,
db=0,
password='encryptedpasswordstring',
ssl=True
)
4. Implementando as Rotas da Aplicação
Rota da Página Inicial
A rota da página inicial renderiza a página principal da aplicação:
.route('/')
def index():
return render_template('index.html')
Consulta de Faixa de População com Cache no Redis
Esta rota lida com consultas de faixas de população. Primeiro verifica se o resultado está em cache no Redis. Se não estiver, faz a consulta no banco de dados e armazena o resultado para uso futuro:
.route('/population-range', methods=['GET', 'POST'])
def population_range():
if request.method == 'POST': # input params defined for this api
year = request.form['yr1']
range1_start = request.form['r1']
range1_end = request.form['r2']
range2_start = request.form['r3']
range2_end = request.form['r4']
range3_start = request.form['r5']
range3_end = request.form['r6']
# Map year to column name
year_map = {
'2010': 'ten',
'2011': 'eleven',
'2012': 'twelve',
'2013': 'thirteen',
'2014': 'fourteen',
'2015': 'fifteen',
'2016': 'sixteen',
'2017': 'seventeen',
'2018': 'eighteen'
}
year_column = year_map.get(year, 'ten') # Default to 'ten' if year not found
# Build cache key
cache_key = f"population_range_{year_column}_{range1_start}_{range1_end}_{range2_start}_{range2_end}_{range3_start}_{range3_end}"
# Check if result is cached
cached_result = redis_client.get(cache_key)
if cached_result:
result = eval(cached_result) # Deserialize cached result
time_taken = 0 # No database query, so time taken is negligible
cache_status = "Cache Hit"
else:
# Build SQL query
query = f"""
SELECT
SUM(CASE WHEN {year_column} BETWEEN '{range1_start}' AND '{range1_end}' THEN 1 ELSE 0 END) AS range1_count,
SUM(CASE WHEN {year_column} BETWEEN '{range2_start}' AND '{range2_end}' THEN 1 ELSE 0 END) AS range2_count,
SUM(CASE WHEN {year_column} BETWEEN '{range3_start}' AND '{range3_end}' THEN 1 ELSE 0 END) AS range3_count
FROM popul
"""
print(query) # For debugging
# Execute query and measure time
start_time = time()
result = engine.execute(query).fetchall()
end_time = time()
time_taken = end_time - start_time
cache_status = "Cache Miss"
# Cache the result
redis_client.set(cache_key, str(result), ex=3600) # Cache for 1 hour
return render_template('display.html', data1=result, t1=time_taken, cache_status=cache_status)
return render_template('index.html')
Buscar Condados por Estado com Cache no Redis
Esta rota recupera condados para um determinado código de estado. Também utiliza o Redis para armazenar em cache os resultados:
.route('/counties-by-state', methods=['GET', 'POST'])
def counties_by_state():
if request.method == 'POST':
state_code = request.form['state_code']
# Build cache key
cache_key = f"counties_by_state_{state_code}"
# Check if result is cached
cached_result = redis_client.get(cache_key)
if cached_result:
result = eval(cached_result) # Deserialize cached result
time_taken = 0 # No database query, so time taken is negligible
cache_status = "Cache Hit"
else:
# Build SQL query
query = f"""
SELECT county
FROM dbo.county
WHERE state = (SELECT state FROM codes WHERE code = '{state_code}')
"""
print(query) # For debugging
# Execute query and measure time
start_time = time()
result = engine.execute(query).fetchall()
end_time = time()
time_taken = end_time - start_time
cache_status = "Cache Miss"
# Cache the result
redis_client.set(cache_key, str(result), ex=3600) # Cache for 1 hour
return render_template('counties.html', data=result, time_taken=time_taken, cache_status=cache_status)
return render_template('index.html')
Recuperar Estados por Faixa de População com Cache no Redis
Esta rota busca estados dentro de uma faixa de população especificada e armazena em cache os resultados:
.route('/states-by-population', methods=['GET', 'POST'])
def states_by_population():
if request.method == 'POST':
year = request.form['year']
population_start = request.form['population_start']
population_end = request.form['population_end']
# Map year to column name
year_map = {
'2010': 'ten',
'2011': 'eleven',
'2012': 'twelve',
'2013': 'thirteen',
'2014': 'fourteen',
'2015': 'fifteen',
'2016': 'sixteen',
'2017': 'seventeen',
'2018': 'eighteen'
}
year_column = year_map.get(year, 'ten') # Default to 'ten' if year not found
# Build cache key
cache_key = f"states_by_population_{year_column}_{population_start}_{population_end}"
# Check if result is cached
cached_result = redis_client.get(cache_key)
if cached_result:
result = eval(cached_result) # Deserialize cached result
time_taken = 0 # No database query, so time taken is negligible
cache_status = "Cache Hit"
else:
# Build SQL query
query = f"""
SELECT state
FROM popul
WHERE {year_column} BETWEEN '{population_start}' AND '{population_end}'
"""
print(query) # For debugging
# Execute query and measure time
start_time = time()
result = engine.execute(query).fetchall()
end_time = time()
time_taken = end_time - start_time
cache_status = "Cache Miss"
# Cache the result
redis_client.set(cache_key, str(result), ex=3600) # Cache for 1 hour
return render_template('states.html', data=result, time_taken=time_taken, cache_status=cache_status)
return render_template('index.html')
Comparação de Desempenho: SQL Server vs. Redis
| Query Type | Redis Fetch Time | SQL Execution Time |
|---|---|---|
| Consulta de Faixa de População (Em Cache) | 0,002 segundos | 0,000 segundos |
| Consulta de Faixa de População (Nova) | 0,002 segundos | 1,342 segundos |
Principais pontos: O Redis reduz o tempo de execução de ~1,3 segundos para ~0,002 segundos, tornando as consultas 650 vezes mais rápidas!
Como o Redis Melhora o Desempenho
O Redis é um armazenamento de dados em memória que atua como uma camada de cache entre a aplicação e o banco de dados. Veja como ele funciona em nossa aplicação:
- Chave de Cache. Uma chave única é gerada para cada consulta com base em seus parâmetros.
- Verificação de Cache. Antes de executar uma consulta ao banco de dados, a aplicação verifica se o resultado já está em cache no Redis.
- Acerto de Cache. Se o resultado for encontrado no Redis, ele é retornado imediatamente, evitando uma consulta ao banco de dados.
- Erro de Cache. Se o resultado não for encontrado, a consulta é executada, e o resultado é armazenado em cache no Redis para uso futuro.
- Expiração de Cache. Os resultados em cache são definidos para expirar após um tempo especificado (por exemplo, 1 hora) para garantir a frescura dos dados.
Ao armazenar em cache resultados de consultas frequentemente acessadas, o Redis reduz significativamente a carga no banco de dados e melhora os tempos de resposta para consultas repetidas.
Conclusão
Neste artigo, construímos uma aplicação Flask que interage com um banco de dados SQL Server para analisar dados populacionais. Integramos o Redis para armazenar em cache resultados de consultas, melhorando o desempenho e reduzindo a carga no banco de dados. Seguindo as melhores práticas, você pode estender esta aplicação para lidar com consultas mais complexas e escalá-la para uso em produção.
Link: O código-fonte desta aplicação completa pode ser encontrado no GitHub.
Source:
https://dzone.com/articles/build-data-analytics-platform-flask-sql-redis