st.connection con SQL, Snowflake y st.secrets para credenciales

st.secrets: gestión segura de credenciales

La gestión de credenciales es un aspecto crítico en cualquier aplicación que conecta con bases de datos o APIs externas. Las credenciales nunca deben estar en el código fuente ni versionarse en el repositorio, ya que cualquier filtración podría comprometer los sistemas de producción.

st.secrets resuelve este problema leyendo las credenciales del archivo .streamlit/secrets.toml, que debe añadirse al .gitignore para evitar su publicación accidental. Este archivo utiliza el formato TOML (Tom's Obvious, Minimal Language), que permite organizar las credenciales en secciones con una sintaxis clara y legible:

# .streamlit/secrets.toml (añadir al .gitignore)

# Credenciales simples
db_password = "mi-contraseña-segura"
api_key = "sk-1234567890abcdef"

# Conexión SQL
[connections.postgres]
type = "sql"
url = "postgresql://usuario:contraseña@localhost:5432/mi_base_datos"

[connections.mysql]
type = "sql"
url = "mysql+pymysql://usuario:contraseña@localhost:3306/mi_base_datos"

[connections.sqlite]
type = "sql"
url = "sqlite:///./datos_locales.db"

# Snowflake
[connections.snowflake]
account = "mi-cuenta.snowflakecomputing.com"
user = "usuario"
password = "contraseña"
warehouse = "COMPUTE_WH"
database = "MI_BD"
schema = "PUBLIC"
role = "SYSADMIN"

Acceder a los secretos en el código:

import streamlit as st

# Acceso por clave
api_key = st.secrets["api_key"]
db_password = st.secrets["db_password"]

# Acceso con notación de punto
password = st.secrets.db_password

# Secciones anidadas
db_url = st.secrets["connections"]["postgres"]["url"]

flowchart TD
    A[App Streamlit] --> B[st.connection nombre tipo]
    B --> C{"Tipo de fuente?"}
    C -->|sql| D[SQLAlchemy URL]
    C -->|snowflake| E[Conector oficial]
    C -->|gsheets| F[Google Sheets]
    C -->|custom| G[Subclase BaseConnection]
    D --> H[conn.query SELECT cached por TTL]
    E --> I[snow.query con sessión]
    F --> J[Lee hoja como DataFrame]
    G --> K[Implementación propia]
    H --> L[DataFrame disponible]
    I --> L
    J --> L
    M[secrets.toml] --> N[Credenciales seguras fuera de git]
    N --> B
    L --> O[Dashboard con datos vivos]

st.connection con SQL (SQLAlchemy)

st.connection es la forma recomendada de establecer conexiones a bases de datos en Streamlit. Gestiona automáticamente el pool de conexiones, la reconexión ante fallos y la integración con el sistema de secretos. La conexión se crea una sola vez y se reutiliza en todas las re-ejecuciones del script:

import streamlit as st
import pandas as pd

# Crear conexión (usa la configuración de secrets.toml)
conn = st.connection("postgres", type="sql")

# Consulta simple con caché de 10 minutos
df = conn.query("SELECT * FROM ventas WHERE año = 2026", ttl=600)
st.dataframe(df, use_container_width=True)

# Consulta con parámetros
region = st.selectbox("Región", ["Norte", "Sur", "Este", "Oeste"])
df_region = conn.query(
    "SELECT * FROM ventas WHERE region = :region ORDER BY fecha DESC",
    params={"region": region},
    ttl=300
)
st.dataframe(df_region)

# Invalidar caché manualmente
if st.button("Actualizar datos"):
    st.cache_data.clear()
    st.rerun()

SQLite local para desarrollo y prototipado

Para desarrollo local o aplicaciones embebidas, SQLite no requiere servidor:

import streamlit as st
import sqlite3
import pandas as pd

# Crear base de datos SQLite local si no existe
def inicializar_bd():
    conn = sqlite3.connect("datos.db")
    conn.execute("""
        CREATE TABLE IF NOT EXISTS productos (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            nombre TEXT NOT NULL,
            precio REAL,
            stock INTEGER
        )
    """)
    conn.execute("INSERT OR IGNORE INTO productos VALUES (1, 'Laptop', 1299.99, 25)")
    conn.execute("INSERT OR IGNORE INTO productos VALUES (2, 'Monitor', 349.50, 12)")
    conn.commit()
    conn.close()

inicializar_bd()

# Usar st.connection con SQLite
conn = st.connection("sqlite", type="sql")
df = conn.query("SELECT * FROM productos", ttl=60)
st.dataframe(df, use_container_width=True)

Conexión directa con psycopg2 y session_state

Para operaciones de escritura (INSERT, UPDATE, DELETE) que conn.query() no gestiona directamente:

import streamlit as st
import psycopg2
import pandas as pd

@st.cache_resource
def obtener_conexion():
    return psycopg2.connect(
        host=st.secrets["connections"]["postgres"]["host"],
        port=st.secrets["connections"]["postgres"]["port"],
        dbname=st.secrets["connections"]["postgres"]["dbname"],
        user=st.secrets["connections"]["postgres"]["user"],
        password=st.secrets["connections"]["postgres"]["password"]
    )

conn = obtener_conexion()

# Lectura
@st.cache_data(ttl=300)
def cargar_clientes():
    with conn.cursor() as cur:
        cur.execute("SELECT id, nombre, email, activo FROM clientes ORDER BY nombre")
        cols = [desc[0] for desc in cur.description]
        return pd.DataFrame(cur.fetchall(), columns=cols)

df = cargar_clientes()
st.dataframe(df, use_container_width=True)

# Escritura (INSERT)
st.subheader("Añadir nuevo cliente")
with st.form("nuevo_cliente"):
    nombre = st.text_input("Nombre *")
    email = st.text_input("Email *")
    enviado = st.form_submit_button("Guardar")

    if enviado and nombre and email:
        with conn.cursor() as cur:
            cur.execute("INSERT INTO clientes (nombre, email) VALUES (%s, %s)", (nombre, email))
        conn.commit()
        st.cache_data.clear()  # Invalidar caché para recargar la tabla
        st.success(f"Cliente '{nombre}' añadido correctamente.")
        st.rerun()

Snowflake con conector oficial

import streamlit as st

# Requiere: pip install snowflake-connector-python
conn = st.connection("snowflake")

df = conn.query(
    "SELECT * FROM VENTAS_2026 WHERE REGION = 'ES' LIMIT 1000",
    ttl=3600
)
st.dataframe(df)

# Ejecutar stored procedure
conn.session().sql("CALL ACTUALIZAR_METRICAS()").collect()

El conector oficial de Snowflake para Streamlit (snowflake-connector-python) gestiona automáticamente la autenticación, el pool de conexiones y la compatibilidad con los tipos de datos de Snowflake.

Al desplegar en Streamlit Community Cloud, los secretos se configuran directamente en el panel de la aplicación (Settings > Secrets), sin necesidad del archivo secrets.toml local. La interfaz web acepta el mismo formato TOML y las claves quedan cifradas en los servidores de Streamlit.

Contexto: la filosofía de st.connection

Antes de st.connection (introducido en Streamlit 1.22), los desarrolladores improvisaban conexiones a bases de datos con patrones mixtos: @st.cache_resource alrededor de un cliente SQLAlchemy, funciones utilitarias que leían st.secrets manualmente, y gestión ad-hoc de la invalidación de caché. El resultado era código frágil, poco reutilizable y propenso a fugas de conexiones.

st.connection unifica este patrón en una sola API. Internamente, la función mantiene un pool de conexiones asociado a la sesión, gestiona el TTL de las consultas, expone un método .query() que devuelve DataFrames directamente y se integra con st.secrets para leer la configuración sin código adicional. El resultado: conectar con PostgreSQL requiere 3 líneas y la aplicación es robusta frente a desconexiones, reruns y cambios de código en caliente.

Explicación línea por línea del ejemplo PostgreSQL

1. conn = st.connection("postgres", type="sql") crea (o recupera) la conexión identificada por "postgres" en secrets.toml. El parámetro type="sql" le indica a Streamlit que use el wrapper SQLAlchemy; el pool de conexiones se configura automáticamente.
2. conn.query("SELECT ...", ttl=600) ejecuta la consulta y devuelve un DataFrame. ttl=600 cachea el resultado durante 10 minutos, lo que evita que múltiples usuarios golpeen la base de datos con la misma consulta.
3. Las consultas parametrizadas usan el formato :param y se pasan los valores con params={"region": region}. Esto previene inyección SQL y mejora la cacheabilidad.
4. st.cache_data.clear() invalida todas las cachés de datos cuando el usuario pulsa el botón "Actualizar". Es una operación global, así que en apps grandes conviene usar query.clear() sobre funciones específicas.

Tabla de parámetros de st.connection y conn.query

| Parámetro | Contexto | Descripción | |-----------|----------|-------------| | name | st.connection | Nombre en secrets.toml que identifica la conexión | | type | st.connection | "sql", "snowflake" o cualquier tipo personalizado | | max_entries | st.connection | Límite de instancias cacheadas | | ttl | st.connection / conn.query | Tiempo de vida en segundos de los resultados | | sql | conn.query | Cadena SQL (soporta parámetros :param) | | params | conn.query | Diccionario de parámetros a sustituir | | index_col | conn.query | Columna a usar como índice del DataFrame resultante | | chunksize | conn.query | Si se específica, devuelve un generador de chunks |

Errores comunes

Credenciales versionadas por error. Si olvidas añadir .streamlit/secrets.toml al .gitignore, las credenciales acabarán en el repositorio. Audita con git log --all --source -- .streamlit/secrets.toml antes del primer push.

Formatos de URL incorrectos. Cada driver tiene su propia convención: PostgreSQL usa postgresql://, MySQL suele requerir mysql+pymysql:// para que funcione con SQLAlchemy. Si ves un error del tipo No driver found, revisa el prefijo.

Consultas sin TTL. Si olvidas el parámetro ttl, Streamlit cacheará el resultado indefinidamente durante la sesión. Es peligroso en aplicaciones donde los datos cambian con frecuencia, porque el usuario verá información obsoleta.

Escribir con conn.query(). El método .query() es solo para lectura. Para INSERT, UPDATE o DELETE, necesitas obtener una sesión SQLAlchemy con conn.sessión() o usar un cliente de bajo nivel como psycopg2 con @st.cache_resource.

Fuga de conexiones con clientes nativos. Si usas psycopg2 o mysql-connector-python directamente, recuerda cerrar cursores y gestionar las reconexiones manualmente. st.connection lo hace por ti.

Mejores prácticas

• Usa st.connection siempre que sea posible: es la forma idiomática y más mantenible.
• Mantén los TTL cortos (5-10 minutos) para datos transaccionales y largos (1-2 horas) para dimensiones relativamente estables.
• Pon las consultas pesadas dentro de funciones con @st.cache_data para beneficiarte también del caching por argumentos.
• Documenta en un README del proyecto qué claves debe contener secrets.toml (sin incluir valores) para facilitar el onboarding de otros desarrolladores.
• Para bases de datos de producción, considera crear un usuario de solo lectura dedicado a la app Streamlit y limitar los permisos a las tablas estrictamente necesarias.
• En Snowflake, prefiere SELECT ... LIMIT N durante el desarrollo para no consumir créditos innecesariamente.