En este artículo, repasaré el desarrollo de una aplicación web basada en Flask que interactúa con una base de datos de SQL Server para analizar datos de población. La aplicación permite a los usuarios consultar rangos de población, obtener condados por estado y recuperar estados dentro de rangos de población específicos. También discutiré cómo integrar Redis para almacenar en caché los resultados de consultas y mejorar el rendimiento.
¿Por qué Flask, SQL Server y Redis?
Flask es un marco de trabajo web ligero y flexible en Python que es perfecto para construir aplicaciones web de tamaño pequeño a mediano. Proporciona las herramientas necesarias para crear APIs RESTful, renderizar plantillas HTML dinámicas e interactuar con bases de datos. Por otro lado, SQL Server es un sólido sistema de gestión de bases de datos relacionales (RDBMS) ampliamente utilizado en aplicaciones empresariales. Combinar Flask con SQL Server nos permite construir una aplicación potente para análisis y visualización de datos.
Para mejorar aún más el rendimiento, integraremos Redis, una base de datos en memoria, para almacenar en caché los resultados de consultas frecuentemente accedidos. Esto reduce la carga en la base de datos y acelera los tiempos de respuesta para consultas repetidas.
Resumen de la aplicación
Nuestra aplicación Flask realiza las siguientes tareas:
- Consulta de rangos de población. Los usuarios pueden especificar un año y un rango de población para obtener el conteo de estados que caen dentro de esos rangos.
- Obtener condados por estado. Los usuarios pueden ingresar un código de estado para recuperar una lista de condados.
- Recuperar estados por rango de población. Los usuarios pueden especificar un rango de población y un año para obtener una lista de estados dentro de ese rango.
- Nota. Para probar, siéntase libre de crear su propio esquema en la base de datos e insertar datos de muestra según sea necesario basándose en las siguientes API compartidas utilizando consultas SQL. Además, las páginas HTML que se utilizan aquí pueden ser un diseño de tabla básico que captura los datos devueltos del código de la aplicación Flask y muestra los resultados.
Vamos a entrar en los detalles de la implementación.
Configuración de la Aplicación Flask
1. Requisitos previos
Antes de comenzar, asegúrese de tener lo siguiente instalado a través de su terminal raíz (comandos compatibles con MacOS):
- Python 3.x
- Flask (
pip install flask) - SQLAlchemy (
pip install sqlalchemy) - PyODBC (
pip install pyodbc) - Redis (
pip install redis)
2. Conexión a la Base de Datos
Usamos SQLAlchemy para conectarnos a la base de datos de SQL Server. Aquí se explica cómo se puede configurar la conexión:
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 cadena de conexión utiliza el controlador ODBC para SQL Server e incluye parámetros para encriptación y tiempo de espera.
3. Configuración de Redis
Redis se utiliza para almacenar en caché los resultados de las consultas. Así es como se configura la conexión a Redis:
import redis
# Redis connection
redis_client = redis.StrictRedis(
host='username.redis.cache.windows.net',
port=6380,
db=0,
password='encryptedpasswordstring',
ssl=True
)
4. Implementar las Rutas de la Aplicación
Ruta de la Página de Inicio
La ruta de la página de inicio muestra la página principal de la aplicación:
.route('/')
def index():
return render_template('index.html')
Consulta de Rango de Población con Caché de Redis
Esta ruta maneja consultas para rangos de población. Primero verifica si el resultado está almacenado en Redis. Si no, realiza la consulta a la base de datos y almacena el 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')
Obtener Condados por Estado con Caché de Redis
Esta ruta recupera condados para un código de estado dado. También utiliza Redis para almacenar en caché los 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')
Obtener Estados por Rango de Población con Caché de Redis
Esta ruta obtiene estados dentro de un rango de población especificado y almacena en caché los 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')
Comparación de Rendimiento: SQL Server vs. Redis
| Query Type | Redis Fetch Time | SQL Execution Time |
|---|---|---|
| Consulta de Rango de Población (Caché) | 0.002 segundos | 0.000 segundos |
| Consulta de Rango de Población (Nueva) | 0.002 segundos | 1.342 segundos |
Lección clave: ¡Redis reduce el tiempo de ejecución de ~1.3 segundos a ~0.002 segundos, ¡haciendo las consultas 650 veces más rápidas!
Cómo Redis mejora el rendimiento
Redis es un almacén de datos en memoria que actúa como una capa de caché entre la aplicación y la base de datos. Así es como funciona en nuestra aplicación:
- Clave de caché. Se genera una clave única para cada consulta basada en sus parámetros.
- Comprobación de la caché. Antes de ejecutar una consulta a la base de datos, la aplicación comprueba si el resultado ya está almacenado en Redis.
- Acertar en la caché. Si el resultado se encuentra en Redis, se devuelve de inmediato, evitando una consulta a la base de datos.
- Fallar en la caché. Si el resultado no se encuentra, se ejecuta la consulta y el resultado se almacena en Redis para un uso futuro.
- Caducidad de la caché. Los resultados almacenados en caché se establecen para que caduquen después de un tiempo especificado (por ejemplo, 1 hora) para garantizar la frescura de los datos.
Al almacenar en caché los resultados de consultas frecuentemente accedidas, Redis reduce significativamente la carga en la base de datos y mejora los tiempos de respuesta para las consultas repetidas.
Conclusión
En este artículo, construimos una aplicación Flask que interactúa con una base de datos de SQL Server para analizar datos de población. Integrarnos Redis para almacenar en caché los resultados de las consultas, mejorando el rendimiento y reduciendo la carga de la base de datos. Siguiendo las mejores prácticas, puedes ampliar esta aplicación para manejar consultas más complejas y escalarla para su uso en producción.
Enlace: El código fuente de esta aplicación completa se puede encontrar en GitHub.
Source:
https://dzone.com/articles/build-data-analytics-platform-flask-sql-redis