Streamlit — Módulo 4

Seguridad · Formularios · Fragmentos · Chat

@josephinoo

2026-04-26

🔐

Módulo 4

Seguridad · Formularios · Fragmentos · Chat

Módulo 4. Hoy llevamos la app a producción: proteger credenciales con st.secrets, tres estrategias de login (manual, FastAPI+JWT, streamlit-authenticator), formularios eficientes, fragmentos para re-ejecuciones parciales, e interfaces de chat al estilo ChatGPT.

¿Qué construimos hoy?

📋

Bloque 1 — Formularios

st.form · st.form_submit_button · agrupar inputs sin reruns explosivos

🔐

Bloque 2 — Seguridad y Acceso

st.secrets · login manual · FastAPI + JWT · streamlit-authenticator

⚡

Bloque 3 — @st.fragment

Re-ejecución parcial · aislar secciones pesadas del resto de la app

💬

Bloque 4 — Interfaces de Chat

st.chat_message · st.chat_input · st.write_stream · historial en session_state

Bloque 1 — Formularios (st.form) 📋

El problema: re-ejecuciones explosivas

# Sin st.form — cada widget
# dispara una re-ejecución completa

ciudad = st.selectbox(
    "Ciudad",
    ["Quito", "Guayaquil", "Cuenca"]
)  # ← rerun #1

presupuesto = st.slider(
    "Presupuesto", 50000, 300000
)  # ← rerun #2

habitaciones = st.number_input(
    "Habitaciones", 1, 10
)  # ← rerun #3

# Con 10 filtros = 10 reruns
# Si cada uno consulta la DB... 💥

Re-ejecuciones al interactuar

0

re-ejecuciones

Toca los widgets →

🏙️ Ciudad

💰 Slider

🛏️ Habitaciones

📐 m²

🏊 Amenidades

⚠️ ¡Demasiados reruns! La DB estaría sufriendo...

Demo en vivo: abrir app con 5 widgets sin form, añadir st.write que muestre un contador de ejecuciones. Ver cómo sube en tiempo real al mover cada slider.

La analogía del carrito de compras

❌ Sin formulario

🛒 Metes producto 1 → pasa tarjeta

🛒 Metes producto 2 → pasa tarjeta

🛒 Metes producto 3 → pasa tarjeta

✏️ Cambias cantidad → pasa tarjeta

Cobra 4 veces por hacer UNA compra

VS

✅ Con st.form

🛒 Metes todos los productos en paz

✏️ Ajustas cantidades sin miedo

🎁 Aplicas cupones y descuentos

💳 Un solo clic → "Ir a la caja"

st.form_submit_button = la caja registradora

💡 st.form agrupa todos sus widgets y solo dispara una re-ejecución cuando el usuario presiona st.form_submit_button.

st.form — código completo

import streamlit as st

st.subheader("Buscador de Propiedades")

with st.form("buscador_casas"):
    st.write("Configura tu búsqueda:")
    col1, col2 = st.columns(2)

    ciudad = col1.selectbox(
        "Ciudad",
        ["Quito", "Guayaquil", "Cuenca"]
    )
    presupuesto = col2.slider(
        "Presupuesto Máximo ($)",
        50_000, 300_000, 100_000,
        step=5_000
    )
    habitaciones = st.number_input(
        "Mínimo de habitaciones", 1, 10, 2
    )
    solo_nuevas = st.checkbox(
        "Solo propiedades nuevas"
    )

    # El único botón que dispara el rerun
    enviado = st.form_submit_button(
        "🔍 Buscar Propiedades",
        type="primary"
    )

if enviado:
    st.success(
        f"Buscando en **{ciudad}** · "
        f"≤ ${presupuesto:,} · "
        f"{habitaciones}+ hab."
    )
    # df_filtrado = df[
    #   (df.ciudad == ciudad) &
    #   (df.precio <= presupuesto)
    # ]

📋 Buscador de Propiedades

Ciudad

QuitoGuayaquilCuenca

Presupuesto Máximo ($)

$100,000

Mínimo de habitaciones

2 hab.

⚡ 0 reruns al ajustar · 1 rerun al buscar

Demo en vivo: mostrar el contador con y sin form. El impacto visual es inmediato. El form tiene un key como string que debe ser único en la app.

Formularios Avanzados 📋

Formularios con validación en tiempo real

import streamlit as st
import re

st.subheader("Registro de Cliente")

with st.form("registro_cliente"):
    nombre = st.text_input(
        "Nombre completo",
        placeholder="Ana García"
    )
    email = st.text_input(
        "Email",
        placeholder="ana@empresa.com"
    )
    telefono = st.text_input(
        "Teléfono (opcional)",
        placeholder="+593 99 123 4567"
    )
    presupuesto = st.select_slider(
        "Rango de presupuesto",
        options=["< $50k", "$50k-100k",
                 "$100k-200k", "> $200k"],
        value="$50k-100k"
    )
    acepta_terminos = st.checkbox(
        "Acepto los términos y condiciones"
    )
    enviado = st.form_submit_button(
        "📝 Registrar Cliente",
        type="primary"
    )

if enviado:
    # Validar DESPUÉS de enviar el form
    errores = []

    if not nombre.strip():
        errores.append("El nombre es requerido")

    if not re.match(
        r'^[\w.-]+@[\w.-]+\.\w{2,}$', email
    ):
        errores.append("Email inválido")

    if not acepta_terminos:
        errores.append("Debes aceptar términos")

    if errores:
        for e in errores:
            st.error(f"❌ {e}")
    else:
        st.success(
            f"✅ {nombre} registrado correctamente!"
        )
        guardar_cliente(nombre, email, presupuesto)

⚠️ La validación ocurre DESPUÉS del submit. Dentro del form no hay reruns, así que no puedes mostrar mensajes de error en tiempo real mientras escribe. La validación en el bloque if enviado: es el patrón correcto.

# ── Patrón avanzado: re-mostrar el form
# si hay errores (UX mejorada)

if enviado and not errores:
    st.session_state.registro_ok = True

if st.session_state.get("registro_ok"):
    st.balloons()
    st.success("¡Registro completado!")
    if st.button("Registrar otro cliente"):
        st.session_state.registro_ok = False
        st.rerun()
else:
    # El form se muestra (o re-muestra)
    # Si hubo errores, están visibles
    # justo debajo del formulario
    pass

✅ Buenas prácticas: limpiar el form después del envío exitoso, guardar el estado del éxito en session_state.

La validación después del submit es el patrón correcto en Streamlit. Si se quiere UX más rica con validación en tiempo real, la alternativa es NO usar st.form y en cambio validar en cada rerun — pero eso vuelve al problema de reruns explosivos. Para la mayoría de casos, validar post-submit es suficiente.

st.form con múltiples secciones y columnas

import streamlit as st

st.subheader("Ficha Completa de Inmueble")

with st.form("ficha_inmueble"):

    # ── Sección 1: Identificación ──────────
    st.markdown("#### 📍 Ubicación")
    c1, c2, c3 = st.columns([2, 1, 1])
    ciudad    = c1.selectbox(
        "Ciudad", ["Quito","Guayaquil","Cuenca"]
    )
    sector    = c2.text_input("Sector")
    codigo    = c3.text_input("Código postal")

    st.divider()

    # ── Sección 2: Características ─────────
    st.markdown("#### 🏠 Características")
    c4, c5, c6 = st.columns(3)
    tipo       = c4.selectbox(
        "Tipo", ["Casa","Departamento","Oficina"]
    )
    habitaciones = c5.number_input("Hab.", 1,20,3)
    banos        = c6.number_input("Baños",1,10,2)

    area        = st.slider(
        "Área (m²)", 30, 500, 120
    )

    st.divider()

    # ── Sección 3: Precio y extras ─────────
    st.markdown("#### 💰 Precio y Extras")
    precio     = st.number_input(
        "Precio ($)", 10000, 1000000, 150000,
        step=5000
    )
    extras     = st.multiselect(
        "Amenidades",
        ["Parqueadero","Piscina","Gimnasio",
         "Seguridad 24h","Terraza","Bodega"]
    )
    disponible = st.date_input(
        "Disponible desde"
    )

    # ── Botones de acción ──────────────────
    col_b1, col_b2 = st.columns([1, 3])
    guardar = col_b1.form_submit_button(
        "💾 Guardar", type="primary"
    )
    borrador = col_b2.form_submit_button(
        "📝 Guardar borrador"
    )

if guardar:
    st.success("✅ Inmueble guardado en la DB")
if borrador:
    st.info("📝 Borrador guardado localmente")

📋 Ficha Completa de Inmueble

📍 Ubicación

Quito

🏠 Características

Casa

💰 Precio y Extras

💡 Un form puede tener múltiples form_submit_button. Cada uno retorna True/False independientemente.

Este es un ejemplo realista de formulario de alta. Tres secciones con st.divider entre ellas, múltiples columnas por sección, y dos botones de submit distintos — uno para guardar definitivo y otro para guardar como borrador. El form agrupa todo sin un solo rerun adicional.

st.form + @st.cache_data — el combo perfecto

import streamlit as st
import pandas as pd

@st.cache_data
def cargar_propiedades() -> pd.DataFrame:
    """Carga UNA sola vez — costosa."""
    return pd.read_csv("propiedades_ecuador.csv")

@st.cache_data
def filtrar_propiedades(
    ciudad: str,
    precio_max: int,
    habitaciones: int
) -> pd.DataFrame:
    """Cache por combinación de parámetros."""
    df = cargar_propiedades()
    return df[
        (df["ciudad"] == ciudad) &
        (df["precio"] <= precio_max) &
        (df["habitaciones"] >= habitaciones)
    ]

# ── El form evita reruns al ajustar sliders
with st.form("filtros"):
    ciudad = st.selectbox(
        "Ciudad",
        ["Quito", "Guayaquil", "Cuenca"]
    )
    precio_max     = st.slider(
        "Precio máximo", 50_000, 500_000
    )
    min_hab        = st.number_input("Hab.", 1, 10)
    buscar         = st.form_submit_button(
        "Buscar", type="primary"
    )

# Solo filtra cuando se presiona el botón
# Y si ya buscaste con los mismos parámetros,
# @st.cache_data devuelve el resultado
# sin recalcular → DOBLE OPTIMIZACIÓN
if buscar:
    with st.spinner("Filtrando..."):
        resultados = filtrar_propiedades(
            ciudad, precio_max, min_hab
        )
    st.metric("Propiedades encontradas",
              len(resultados))
    st.dataframe(resultados, use_container_width=True)

🚀 Doble optimización:

1️⃣

st.form → evita reruns al mover sliders. La función de filtrado no se llama mientras el usuario ajusta parámetros.

2️⃣

@st.cache_data → si el usuario busca dos veces con los mismos parámetros, el resultado se devuelve desde la caché instantáneamente.

⚡

Resultado: app ultra-fluida incluso con 100k filas y consultas pesadas a bases de datos.

📊 Para DataFrames grandes: considera también @st.cache_data(ttl=300) para actualizar la caché cada 5 minutos con datos frescos de la DB.

La combinación st.form + @st.cache_data es el patrón de rendimiento estándar para apps con datos. st.form elimina los reruns innecesarios mientras el usuario configura filtros, y @st.cache_data evita recalcular cuando los parámetros ya fueron procesados antes. Para apps con bases de datos reales, esto puede reducir el tiempo de respuesta de 5 segundos a instantáneo.

Bloque 2 — Seguridad y Acceso 🔐

El problema: credenciales en el código

# ❌ NUNCA hagas esto
import psycopg2

conn = psycopg2.connect(
    host="prod.mydb.com",
    user="admin",
    password="sup3r_s3cr3t0",  # 💀
    database="clientes"
)
# Git push → credencial pública
# Bots escanean repos en segundos.

⚠️ El error de seguridad #1 de los data scientists.

💀

Password visible en el código fuente

💀

Git push → credencial comprometida

💀

API key quemada → factura inesperada 💸

💀

Logs de Streamlit Cloud exponen el valor

Preguntar a la clase: ¿alguien ha visto credenciales en un repo público? Es muy común. Hay bots que escanean GitHub constantemente buscando API keys. El primer paso siempre: sacar los secretos del código.

st.secrets — el paso cero

.streamlit/secrets.toml🔒 en .gitignore

# Este archivo NUNCA va al repo

USUARIO_ADMIN = "admin"
CLAVE_ADMIN = "clave_segura"
JWT_SECRET = "mi_firma_jwt"
API_URL = "http://localhost:8000"
OPENAI_KEY = "sk-proj-abc123"

app.py✅ seguro

import streamlit as st

# Acceder con clave de diccionario
clave = st.secrets["CLAVE_ADMIN"]

# O con notación de punto
api_key = st.secrets.OPENAI_KEY

# En Streamlit Cloud: panel de Settings
# → Secrets (mismo formato TOML)

📁 mkdir .streamlit && touch .streamlit/secrets.toml → agregar al .gitignore antes del primer commit

st.secrets lee el archivo automáticamente. En Streamlit Cloud se configuran las variables desde el panel — el código no cambia. Es lo primero que hay que hacer antes de cualquier otra cosa.

Método 1 — Login manual con session_state

import streamlit as st

# Estado inicial
if "autenticado" not in st.session_state:
    st.session_state.autenticado = False

if not st.session_state.autenticado:
    st.title("🔐 Acceso Restringido")
    usuario = st.text_input("Usuario")
    clave   = st.text_input("Contraseña",
                             type="password")

    if st.button("Ingresar", type="primary"):
        if (usuario == st.secrets["USUARIO_ADMIN"]
            and clave == st.secrets["CLAVE_ADMIN"]):
            st.session_state.autenticado = True
            st.rerun()
        else:
            st.error("Credenciales incorrectas")

    st.stop()  # nada más se ejecuta

# ✅ Solo llega aquí si está autenticado
st.title("🏠 Dashboard Principal")
if st.button("Cerrar sesión"):
    st.session_state.autenticado = False
    st.rerun()

🔐

Acceso Restringido

Usuario

Contraseña

❌ Credenciales incorrectas

✅ ¡Bienvenido, admin!

🎯 Ideal para: prototipos, dashboards de un solo administrador, demos rápidas.

El método más simple. Tres piezas: st.secrets para no hardcodear, session_state para recordar entre reruns, y st.stop() para bloquear el resto. Perfecto para apps personales o equipo pequeño.

st.stop() — el guardián del script

import streamlit as st

st.title("App Segura")

if not autenticado:

    mostrar_login()

    st.stop() ← DETIENE AQUÍ

st.subheader("Dashboard")

df = cargar_datos_privados()

st.dataframe(df)

st.plotly_chart(grafico())

1

App inicia → verifica session_state

↓

2

No autenticado → muestra el formulario de login

↓

🛑

st.stop() — ejecución detenida

↓

✗

Dashboard NUNCA se renderiza ni procesa

✅ No es solo visual: la lógica Python posterior tampoco corre. Los datos privados nunca se cargan.

Punto clave: st.stop() no es solo “no mostrar”. El código Python posterior ni siquiera se evalúa. Si tienes una función que conecta a la DB, esa función nunca se llama.

Método 2 — FastAPI como backend de autenticación

🖥️

Streamlit

Frontend

→
POST /token

⚡

FastAPI

Auth Backend

→
verifica

🗄️

Base de Datos

bcrypt hashes

←
JWT token

🔑

JWT

session_state

main.py — FastAPI

from fastapi import FastAPI, Depends, HTTPException
from fastapi.security import OAuth2PasswordRequestForm
import jwt, bcrypt
from datetime import datetime, timedelta

app = FastAPI()
SECRET = "mi_firma_jwt"

USERS_DB = {
    "ana": bcrypt.hashpw(b"pass123",
                          bcrypt.gensalt())
}

@app.post("/token")
async def login(
    form: OAuth2PasswordRequestForm = Depends()
):
    stored = USERS_DB.get(form.username)
    if not stored or not bcrypt.checkpw(
        form.password.encode(), stored
    ):
        raise HTTPException(401,
                            "Credenciales inválidas")

    token = jwt.encode({
        "sub": form.username,
        "exp": datetime.utcnow()
               + timedelta(hours=8)
    }, SECRET, algorithm="HS256")

    return {"access_token": token,
            "token_type":   "bearer"}

app.py — Streamlit

import streamlit as st
import requests, jwt

API_URL = st.secrets["API_URL"]
SECRET  = st.secrets["JWT_SECRET"]

if "token" not in st.session_state:
    st.session_state.token = None

if not st.session_state.token:
    st.title("🔐 Login")
    user = st.text_input("Usuario")
    pw   = st.text_input("Contraseña",
                          type="password")

    if st.button("Ingresar", type="primary"):
        r = requests.post(
            f"{API_URL}/token",
            data={"username": user,
                  "password": pw}
        )
        if r.status_code == 200:
            st.session_state.token = \
                r.json()["access_token"]
            st.rerun()
        else:
            st.error("Credenciales inválidas")
    st.stop()

# Decodificar token y verificar expiración
try:
    payload = jwt.decode(
        st.session_state.token,
        SECRET, algorithms=["HS256"]
    )
except jwt.ExpiredSignatureError:
    st.session_state.token = None
    st.warning("Sesión expirada")
    st.rerun()

st.title(f"Hola, {payload['sub']} 👋")

Este patrón es el que se usa en producción real. FastAPI maneja la autenticación, bcrypt hashea contraseñas, JWT crea tokens sin estado. Streamlit solo hace POST al endpoint y guarda el token. Los datos nunca salen del backend sin un token válido. Ideal cuando ya tienes o vas a tener una API que sirva a múltiples frontends.

Método 3 — streamlit-authenticator 🏆

✅ Login UI integrada 🍪 Cookies persistentes 🔑 Reset contraseña 📧 Recuperar usuario 👤 Registro de usuarios 🔒 bcrypt nativo

import streamlit as st
import streamlit_authenticator as stauth

# 1. Definir credenciales (hashes bcrypt)
config = {
    "credentials": {
        "usernames": {
            "ana": {
                "name":     "Ana García",
                "email":    "ana@empresa.com",
                "password": "$2b$12$..."  # hash
            },
            "carlos": {
                "name":     "Carlos López",
                "email":    "carlos@empresa.com",
                "password": "$2b$12$..."  # hash
            }
        }
    },
    "cookie": {
        "name":        "app_cookie",
        "key":         st.secrets["JWT_SECRET"],
        "expiry_days": 30
    }
}

# 2. Crear el autenticador
auth = stauth.Authenticate(
    config["credentials"],
    config["cookie"]["name"],
    config["cookie"]["key"],
    config["cookie"]["expiry_days"],
    auto_hash=True  # hashea automáticamente
)

# 3. Renderizar widget de login
auth.login()

# 4. Reaccionar al estado
if st.session_state["authentication_status"]:
    auth.logout("Cerrar sesión", "sidebar")
    nombre = st.session_state["name"]
    st.title(f"¡Hola, {nombre}! 👋")
    st.write("Aquí va el dashboard...")

elif st.session_state["authentication_status"] \
     is False:
    st.error("Usuario o contraseña incorrectos")

elif st.session_state["authentication_status"] \
     is None:
    st.warning("Ingresa tus credenciales")

🌐 localhost:8501

A

Mi Aplicación

Username

Password

Forgot password? Forgot username?

✅ ¡Bienvenida, Ana García!

$ pip install streamlit-authenticator

📖 Docs: streamlit-authenticator.readthedocs.io

streamlit-authenticator es la solución más completa para producción sin mantener tu propio backend. Tiene login, logout, registro, recuperación de contraseña y usuario, todo con UI integrada. Cookies para persistir la sesión entre visitas. La documentación oficial está en streamlit-authenticator.readthedocs.io.

Hashear contraseñas con stauth.Hasher

import streamlit_authenticator as stauth

# ── Generar hashes (solo una vez al configurar)
hashed = stauth.Hasher(
    ["pass123", "secreto456"]
).generate()
# → ['$2b$12$abc...', '$2b$12$xyz...']

# ── Verificar si un string ya es hash
es_hash = stauth.Hasher.is_hash(
    "$2b$12$abc..."
)  # → True

# ── con auto_hash=True en Authenticate,
# puedes pasar texto plano directamente
# y se hashea automáticamente al inicio.

🔑 Guarda los hashes en secrets.toml. Nunca almacenes el texto plano de las contraseñas.

.streamlit/secrets.toml

[credentials.usernames.ana]
name = "Ana García"
email = "ana@empresa.com"
password = "$2b$12$abc..."

[credentials.usernames.carlos]
name = "Carlos López"
email = "carlos@empresa.com"
password = "$2b$12$xyz..."

[cookie]
name = "mi_app_cookie"
key = "secreto_firma"
expiry_days = 30

El flujo de setup: 1) definir usuarios con contraseñas en texto plano, 2) usar Hasher para obtener los hashes bcrypt, 3) guardar esos hashes en secrets.toml. Con auto_hash=True, streamlit-authenticator hace el paso 2 y 3 automáticamente si pasas texto plano.

Los 3 métodos — ¿cuál usar?

Método	Complejidad	Multi-usuario	Cookies	Ideal para
session_state manual	⭐ Baja	❌ No	❌ No	Prototipos, un solo admin, demos
FastAPI + JWT	⭐⭐⭐ Alta	✅ Sí	⚠️ Manual	Arquitecturas multi-app, API propia, control total
streamlit-authenticator	⭐⭐ Media	✅ Sí	✅ Nativo	Apps Streamlit multi-usuario, sin backend propio

💡 Regla práctica: empieza con session_state para prototipos → escala a streamlit-authenticator para multi-usuario → usa FastAPI + JWT si necesitas una API que sirva a múltiples aplicaciones.

No hay una respuesta correcta única. La complejidad tiene que estar justificada por los requisitos. Para el reto del curso, streamlit-authenticator es el punto dulce entre simplicidad y capacidad real.

Seguridad Avanzada 🛡️

Control de roles (RBAC) con session_state

import streamlit as st
import streamlit_authenticator as stauth

# Definir usuarios con roles
config = {
    "credentials": {
        "usernames": {
            "ana_admin": {
                "name": "Ana García",
                "email": "ana@empresa.com",
                "password": "$2b$12$...",
                "role": "admin"
            },
            "carlos_viewer": {
                "name": "Carlos López",
                "email": "carlos@empresa.com",
                "password": "$2b$12$...",
                "role": "viewer"
            }
        }
    },
    "cookie": {
        "name": "app_cookie",
        "key":  st.secrets["JWT_SECRET"],
        "expiry_days": 30
    }
}

auth = stauth.Authenticate(
    config["credentials"],
    config["cookie"]["name"],
    config["cookie"]["key"],
    config["cookie"]["expiry_days"]
)
auth.login()

if st.session_state["authentication_status"]:
    user = st.session_state["username"]
    role = config["credentials"]["usernames"] \
               [user]["role"]

    st.session_state["role"] = role
    auth.logout("Salir", "sidebar")
    st.title(
      f"Hola, {st.session_state['name']} 👋"
    )

    # ── Control de acceso por rol ──────────
    if role == "admin":
        st.success("🔑 Modo Administrador")
        tab1, tab2, tab3 = st.tabs(
            ["📊 Dashboard",
             "⚙️ Gestión",
             "🗄️ Datos brutos"]
        )
    elif role == "viewer":
        st.info("👁️ Modo Solo Lectura")
        tab1, = st.tabs(["📊 Dashboard"])
        # tab2 y tab3 no existen para viewer

Función de decorador para proteger páginas

def require_role(roles: list):
    """Decorador para proteger secciones
    según el rol del usuario actual."""
    def decorator(func):
        def wrapper(*args, **kwargs):
            current_role = st.session_state.get(
                "role", None
            )
            if current_role not in roles:
                st.error(
                    f"🚫 Acceso denegado. "
                    f"Necesitas rol: {roles}"
                )
                st.stop()
            return func(*args, **kwargs)
        return wrapper
    return decorator

# Uso en cualquier función de la app:
@require_role(["admin"])
def panel_administracion():
    st.subheader("⚙️ Gestión de usuarios")
    st.dataframe(obtener_todos_los_usuarios())
    if st.button("Eliminar usuario"):
        # solo admins llegan aquí
        pass

@require_role(["admin", "editor"])
def panel_edicion():
    st.subheader("✏️ Editor de contenido")
    # admins y editors llegan aquí

🔑 RBAC (Role-Based Access Control) — el patrón estándar de seguridad en aplicaciones empresariales.

RBAC es fundamental en apps reales. La idea es simple: cada usuario tiene un rol, y cada sección de la app verifica ese rol antes de renderizar. El decorador require_role es un patrón elegante en Python que convierte cualquier función en una función protegida sin repetir lógica de verificación.

Variables de entorno como alternativa a secrets.toml

import streamlit as st
import os

# ── Opción 1: st.secrets (recomendado Streamlit)
clave = st.secrets["CLAVE_ADMIN"]

# ── Opción 2: variables de entorno del SO
# Útil en Docker, CI/CD, servidores Linux
clave = os.environ.get(
    "CLAVE_ADMIN",
    "valor_por_defecto"  # si no existe
)

# ── Opción 3: python-dotenv (local dev)
# pip install python-dotenv
from dotenv import load_dotenv
load_dotenv()  # carga el .env
clave = os.getenv("CLAVE_ADMIN")

# ── Estrategia híbrida (producción real)
def get_secret(key: str) -> str:
    """Busca en st.secrets primero,
    cae a variables de entorno."""
    try:
        return st.secrets[key]
    except (KeyError, FileNotFoundError):
        val = os.environ.get(key)
        if val is None:
            raise ValueError(
                f"Secreto '{key}' no encontrado "
                f"en secrets.toml ni en env vars"
            )
        return val

# Uso transparente
DB_URL  = get_secret("DB_URL")
API_KEY = get_secret("OPENAI_KEY")

.env (local, en .gitignore)

.env🔒 .gitignore

CLAVE_ADMIN=clave_segura
JWT_SECRET=mi_firma_jwt
DB_URL=postgresql://localhost/db
OPENAI_KEY=sk-proj-abc123

1

Local: .env + python-dotenv

2

Streamlit Cloud: panel de Secrets

3

Docker / Kubernetes: -e VAR=value

4

CI/CD (GitHub Actions): repository secrets

La estrategia híbrida es lo que usan los equipos profesionales. En desarrollo local usan .env, en producción en Streamlit Cloud usan el panel de Secrets, en servidores propios usan variables de entorno del sistema. El código no cambia — solo el mecanismo de entrega del secreto.

Middleware de autenticación en FastAPI

from fastapi import FastAPI, Depends, HTTPException
from fastapi.security import HTTPBearer, \
                              HTTPAuthorizationCredentials
from fastapi.middleware.cors import CORSMiddleware
import jwt

app = FastAPI(title="Auth API para Streamlit")

# ── CORS: permitir peticiones desde Streamlit ─
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:8501",
                   "https://mi-app.streamlit.app"],
    allow_methods=["POST", "GET"],
    allow_headers=["*"],
)

security = HTTPBearer()

# ── Dependencia reutilizable ──────────────────
def verificar_token(
    creds: HTTPAuthorizationCredentials
          = Depends(security)
) -> dict:
    try:
        payload = jwt.decode(
            creds.credentials,
            SECRET, algorithms=["HS256"]
        )
        return payload
    except jwt.ExpiredSignatureError:
        raise HTTPException(401, "Token expirado")
    except jwt.InvalidTokenError:
        raise HTTPException(401, "Token inválido")

# ── Endpoints protegidos ──────────────────────
@app.get("/api/propiedades")
async def listar_propiedades(
    payload: dict = Depends(verificar_token)
):
    """Solo accesible con JWT válido"""
    usuario = payload["sub"]
    return {"usuario": usuario,
            "datos": obtener_propiedades()}

@app.get("/api/admin/usuarios")
async def listar_usuarios(
    payload: dict = Depends(verificar_token)
):
    """Solo para admins"""
    if payload.get("role") != "admin":
        raise HTTPException(403, "Sin permisos")
    return obtener_todos_los_usuarios()

# ── En Streamlit: llamar endpoints protegidos ─
import streamlit as st
import requests

def api_get(endpoint: str) -> dict:
    """Helper para llamadas autenticadas a API"""
    token = st.session_state.get("token")
    if not token:
        st.error("No autenticado")
        st.stop()

    headers = {"Authorization": f"Bearer {token}"}
    r = requests.get(
        f"{st.secrets['API_URL']}{endpoint}",
        headers=headers
    )

    if r.status_code == 401:
        # Token expirado — forzar re-login
        st.session_state.token = None
        st.warning("Sesión expirada. Re-inicia.")
        st.rerun()

    if r.status_code == 403:
        st.error("🚫 Sin permisos para esta acción")
        st.stop()

    r.raise_for_status()
    return r.json()

# Uso en cualquier parte de la app:
propiedades = api_get("/api/propiedades")
st.dataframe(propiedades["datos"])

🏗️ Con este patrón, Streamlit es solo el frontend. La lógica de negocio y los datos viven en FastAPI.

El middleware de CORS es imprescindible cuando Streamlit y FastAPI corren en dominios distintos. La dependencia verificar_token es el corazón del sistema — se inyecta en cada endpoint protegido. En el lado Streamlit, el helper api_get centraliza el manejo de errores 401/403 para no repetir lógica.

Bloque 3 — @st.fragment ⚡

Rompiendo la regla de oro del Módulo 1

📖 Módulo 1: "Cada interacción re-ejecuta todo el script de arriba a abajo."

❌ Sin fragment

📊 Gráfico pesado (3s)

📈 Tabla grande (2s)

👍 Botón "Like"

✅ Con @st.fragment

📊 Gráfico (congelado ✓)

📈 Tabla (congelada ✓)

⚡ Botón "Like" (solo esto)

import streamlit as st
import time

# Sección pesada — solo carga al inicio
st.subheader("Análisis Inmobiliario")
with st.spinner("Cargando gráfico..."):
    time.sleep(2)  # simula carga pesada
    st.line_chart({"precio": [100,120,95,
                               110,130]})

# ─────────────────────────────────────────
# @st.fragment → solo esta función se
# re-ejecuta al interactuar DESDE adentro
# ─────────────────────────────────────────
@st.fragment
def panel_rapido():
    st.divider()
    favorito = st.toggle("⭐ Guardar favorito")
    if st.button("📤 Compartir análisis"):
        st.success("¡Enlace copiado!")
    if favorito:
        st.info("Guardado en tus favoritos")

panel_rapido()
# El gráfico de arriba NO parpadea
# al interactuar con este panel

@st.fragment redefine parcialmente el scripting flow. Cuando el usuario interactúa con un widget DENTRO de un fragmento, solo ese fragmento se re-ejecuta. El resto queda intacto. Caso de uso clásico: gráfico pesado arriba, widgets ligeros abajo.

@st.fragment — el impacto visual

⏱️ Esta parte tardó 3s en cargar...

📊 [ gráfico pesado renderizado ] ← intacto

📈 [ tabla de 50k filas ] ← intacta

@st.fragment — re-ejecución aislada

Panel de acciones rápidas

1

App inicia → todo se ejecuta una vez (3s)

↓ usuario hace clic en "Guardar favorito"

⚡

Solo panel_rapido() se re-ejecuta (instantáneo)

✗

El gráfico y la tabla NO se vuelven a cargar

⚡ Ideal cuando tienes secciones "pesadas" y widgets "ligeros" conviviendo en la misma pantalla.

📌 @st.fragment(run_every=5) → auto-actualiza el fragmento cada 5 segundos. Perfecto para live data.

El efecto de demostración es clave: mostrar en la app real que el spinner del gráfico pesado NO aparece al interactuar con el fragmento. Eso es lo que convence a los estudiantes. También mencionar run_every para dashboards con datos en tiempo real.

@st.fragment Avanzado ⚡

run_every — dashboards en tiempo real

import streamlit as st
import pandas as pd
import random
import time

st.title("📡 Monitor de Ventas en Vivo")

# ── Esta parte carga una vez (pesada) ─────
@st.cache_data
def cargar_historico():
    return pd.read_csv("ventas_historicas.csv")

df_hist = cargar_historico()
st.line_chart(df_hist.set_index("fecha"))
st.caption("📊 Histórico — actualiza al inicio")

# ── Este fragmento se auto-actualiza ──────
@st.fragment(run_every=5)  # cada 5 segundos
def kpis_en_vivo():
    """Re-ejecuta solo este bloque cada 5s.
    El gráfico histórico de arriba NO recarga."""
    # Simula datos en tiempo real
    ventas_hoy  = random.randint(80, 150)
    ticket_prom = random.uniform(45.0, 95.0)
    activos     = random.randint(12, 35)

    col1, col2, col3 = st.columns(3)
    col1.metric(
        "💰 Ventas hoy",
        f"${ventas_hoy:,}",
        delta=f"{random.randint(-5,15)}%"
    )
    col2.metric(
        "🎫 Ticket promedio",
        f"${ticket_prom:.2f}",
        delta=f"{random.randint(-3,8):.1f}%"
    )
    col3.metric(
        "👥 Usuarios activos",
        activos,
        delta=random.randint(-3, 5)
    )
    st.caption(
        f"⏱️ Actualizado: {time.strftime('%H:%M:%S')}"
    )

kpis_en_vivo()

KPIs en Vivo — actualización cada 5s

$127

Ventas hoy

▲ +8%

$67.40

Ticket prom.

▲ +3.2%

23

Activos

▼ -2

⏱️ Actualizado: --:--:--

📡 run_every=5 acepta segundos (int), timedelta, o strings como "5s", "1m". El fragmento se re-ejecuta automáticamente sin que el usuario haga nada.

run_every es la feature más poderosa de @st.fragment para dashboards operacionales. Los KPIs se actualizan automáticamente cada 5 segundos sin recargar el gráfico histórico pesado de arriba. Esto permite tener una app que parece “viva” sin el costo de recargar toda la pantalla.

Múltiples fragmentos independientes

import streamlit as st

st.title("🏢 Dashboard Ejecutivo")

# Barra de navegación — carga una vez
seccion = st.selectbox(
    "Ver sección:",
    ["Ventas", "Inventario", "Clientes"]
)

# ── Fragmento 1: KPIs (actualiza cada 10s)
@st.fragment(run_every=10)
def bloque_kpis():
    st.subheader("📊 KPIs en vivo")
    col1, col2 = st.columns(2)
    col1.metric("Ventas", "$12,400", "+8%")
    col2.metric("Leads", "47", "+3")

# ── Fragmento 2: Chat lateral (independiente)
@st.fragment
def bloque_chat():
    st.subheader("💬 Asistente")
    if "chat_msgs" not in st.session_state:
        st.session_state.chat_msgs = []
    for m in st.session_state.chat_msgs:
        with st.chat_message(m["role"]):
            st.write(m["content"])
    if q := st.chat_input("Pregunta algo..."):
        st.session_state.chat_msgs.append(
            {"role": "user", "content": q}
        )
        st.session_state.chat_msgs.append(
            {"role": "assistant",
             "content": f"Analizando: {q}"}
        )
        st.rerun(scope="fragment")

# ── Layout: 2 columnas con fragmentos ─────
col_izq, col_der = st.columns([2, 1])

with col_izq:
    bloque_kpis()
    # aquí va el contenido de 'seccion'
    mostrar_seccion(seccion)

with col_der:
    bloque_chat()

⚡

bloque_kpis → se re-ejecuta cada 10s automáticamente

💬

bloque_chat → solo re-ejecuta cuando el usuario escribe en el chat

✓

mostrar_seccion() → solo re-carga cuando cambia el selectbox

🔁 st.rerun(scope="fragment") — re-ejecuta solo el fragmento actual, no toda la app. Disponible desde Streamlit 1.33+.

🧩 Cada fragmento es independiente. Puedes tener 10 fragmentos en la misma app, cada uno con su propio ciclo de vida.

La clave aquí es st.rerun(scope=“fragment”) — re-ejecuta solo el fragmento actual en lugar de toda la app. Esto es crucial para el chat dentro de un fragmento, porque sin scope=“fragment”, el chat dispararía un rerun completo que resetearía el selectbox de arriba.

Bloque 4 — Interfaces de Chat 💬

Los componentes nativos de chat

Componente	Qué hace
st.chat_message	Renderiza una burbuja con avatar (user o assistant). Dentro puedes usar cualquier elemento de Streamlit.
st.chat_input	Barra fija en la parte inferior. Devuelve el texto al presionar Enter o el ícono de envío.
st.write_stream	Consume un generador y muestra el texto palabra a palabra (efecto typewriter / ChatGPT).
session_state	Guarda el historial de mensajes para que no se borre en cada rerun.

🔑 Sin session_state, el chat olvida toda la conversación cada vez que el usuario escribe una letra.

💬 Asistente Inmobiliario

🤖

¡Hola! Soy tu asesor inmobiliario. ¿Qué tipo de propiedad buscas hoy?

Conectar con el Módulo 3: session_state es lo que estudiamos para guardar estado entre reruns. Ahora lo usamos para guardar el historial. Los componentes de chat son solo la presentación visual — la inteligencia real viene de una API de LLM.

Chat con historial persistente

import streamlit as st

st.subheader("🏡 Asistente Inmobiliario")

# ── Inicializar historial ─────────────────
if "mensajes" not in st.session_state:
    st.session_state.mensajes = [{
        "role":    "assistant",
        "content": "¡Hola! Soy tu asesor "
                   "inmobiliario virtual. "
                   "¿Qué tipo de propiedad "
                   "buscas hoy?"
    }]

# ── Renderizar historial completo ─────────
for msg in st.session_state.mensajes:
    with st.chat_message(msg["role"]):
        st.write(msg["content"])

# ── Capturar nueva pregunta ───────────────
pregunta = st.chat_input(
    "Ej: Busco un depto de 2 hab en Quito..."
)

if pregunta:
    # 1. Guardar y mostrar mensaje del usuario
    st.session_state.mensajes.append(
        {"role": "user", "content": pregunta}
    )
    with st.chat_message("user"):
        st.write(pregunta)

    # 2. Generar y mostrar respuesta
    with st.chat_message("assistant"):
        respuesta = st.write_stream(
            generar_respuesta(pregunta)
        )

    # 3. Guardar respuesta en historial
    st.session_state.mensajes.append(
        {"role": "assistant",
         "content": respuesta}
    )

import time

# ── Generador para efecto streaming ──────
def generar_respuesta(pregunta: str):
    """Simula respuesta palabra a palabra"""
    texto = (
        f"Basándome en tu búsqueda de "
        f"'{pregunta}', encontré 3 opciones "
        f"disponibles. La primera está en el "
        f"norte de la ciudad con 2 dormitorios "
        f"y parqueadero incluido..."
    )
    for palabra in texto.split():
        yield palabra + " "
        time.sleep(0.04)  # efecto typewriter

# ── Con OpenAI API real ───────────────────
# from openai import OpenAI
# client = OpenAI(
#     api_key=st.secrets["OPENAI_KEY"]
# )
#
# def generar_respuesta(pregunta):
#     stream = client.chat.completions.create(
#         model="gpt-4o",
#         stream=True,
#         messages=st.session_state.mensajes
#     )
#     for chunk in stream:
#         delta = chunk.choices[0].delta
#         if delta.content:
#             yield delta.content

$ pip install openai

✨ st.write_stream() acepta cualquier generador Python. Compatible directamente con stream=True de OpenAI, Anthropic y Gemini.

El patrón de los tres pasos siempre es el mismo: 1) renderizar historial guardado, 2) capturar input del usuario, 3) agregar mensajes al historial y generar respuesta. La clave es que el historial en session_state sobrevive a los reruns. El generador de Python se puede reemplazar por cualquier API de LLM que soporte streaming.

Chat e Integración con LLMs 💬

Contexto del sistema y memoria de la conversación

import streamlit as st

SYSTEM_PROMPT = """Eres un asesor inmobiliario
experto en el mercado ecuatoriano. Tienes acceso
a una base de datos de propiedades en Quito,
Guayaquil y Cuenca. Responde siempre en español,
de manera concisa y profesional. Si el usuario
pregunta por propiedades, pide sus filtros
(ciudad, presupuesto, habitaciones) antes de
mostrar resultados."""

# ── Inicializar con mensaje del sistema ───
if "historial" not in st.session_state:
    st.session_state.historial = [
        {"role": "system",
         "content": SYSTEM_PROMPT}
    ]
    # El sistema NO se muestra en la UI

# ── Renderizar solo mensajes user/assistant ─
for msg in st.session_state.historial:
    if msg["role"] == "system":
        continue  # ocultar prompt del sistema
    with st.chat_message(msg["role"]):
        st.markdown(msg["content"])

# ── Input ─────────────────────────────────
if prompt := st.chat_input("Escribe aquí..."):
    st.session_state.historial.append(
        {"role": "user", "content": prompt}
    )
    with st.chat_message("user"):
        st.markdown(prompt)

    with st.chat_message("assistant"):
        with st.spinner("Pensando..."):
            respuesta = llamar_llm(
                st.session_state.historial
            )
        st.markdown(respuesta)

    st.session_state.historial.append(
        {"role": "assistant",
         "content": respuesta}
    )

# ── Sidebar: controles del chat ───────────
with st.sidebar:
    st.caption(
        f"💬 {len(st.session_state.historial)-1}"
        f" mensajes en memoria"
    )
    if st.button("🗑️ Limpiar conversación"):
        st.session_state.historial = [
            {"role": "system",
             "content": SYSTEM_PROMPT}
        ]
        st.rerun()

from openai import OpenAI

client = OpenAI(
    api_key=st.secrets["OPENAI_KEY"]
)

def llamar_llm(historial: list) -> str:
    """Envía el historial completo a OpenAI.
    Incluye el system prompt automáticamente."""
    respuesta = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=historial,  # todo el contexto
        max_tokens=500,
        temperature=0.7
    )
    return respuesta.choices[0].message.content

# ── Con streaming (efecto typewriter) ─────
def llamar_llm_stream(historial: list):
    """Generador para st.write_stream()"""
    stream = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=historial,
        max_tokens=500,
        stream=True  # ← activa streaming
    )
    for chunk in stream:
        delta = chunk.choices[0].delta
        if delta.content:
            yield delta.content

🧠 El historial completo (incluyendo el system prompt) se envía en CADA llamada. Los LLMs no tienen memoria propia — toda la memoria vive en session_state.

El system prompt es la instrucción que le da personalidad y contexto al asistente. Se pone al inicio del historial como rol “system” y se envía en cada llamada a la API. El usuario no lo ve. La gestión de memoria es crítica: los LLMs cobran por tokens, así que en conversaciones largas hay que considerar truncar el historial.

Chat multimodal — texto e imágenes

import streamlit as st
import base64
from openai import OpenAI

client = OpenAI(api_key=st.secrets["OPENAI_KEY"])

st.title("🏠 Analiza tu Propiedad con IA")
st.caption(
    "Sube una foto del inmueble y pregunta "
    "lo que quieras."
)

# ── Upload de imagen ──────────────────────
imagen = st.file_uploader(
    "📸 Sube una foto del inmueble",
    type=["jpg", "jpeg", "png", "webp"]
)

if imagen:
    st.image(imagen, caption="Tu propiedad",
             use_container_width=True)

# ── Chat con imagen ───────────────────────
if "msgs_img" not in st.session_state:
    st.session_state.msgs_img = []

for msg in st.session_state.msgs_img:
    with st.chat_message(msg["role"]):
        st.markdown(msg["content"]
                    if isinstance(msg["content"],str)
                    else msg["content"][0]["text"])

pregunta = st.chat_input(
    "Pregunta sobre la foto..."
    if imagen else
    "Sube una imagen primero..."
)

if pregunta and imagen:
    # Convertir imagen a base64
    img_b64 = base64.b64encode(
        imagen.getvalue()
    ).decode()
    img_type = imagen.type

    # Mensaje con imagen + texto
    msg_usuario = {
        "role": "user",
        "content": [
            {"type": "image_url",
             "image_url": {
                 "url": f"data:{img_type};"
                        f"base64,{img_b64}"
             }},
            {"type": "text", "text": pregunta}
        ]
    }
    st.session_state.msgs_img.append(msg_usuario)

    with st.chat_message("user"):
        st.markdown(pregunta)

    with st.chat_message("assistant"):
        respuesta = st.write_stream(
            analizar_imagen(
                st.session_state.msgs_img
            )
        )
    st.session_state.msgs_img.append(
        {"role": "assistant",
         "content": respuesta}
    )

def analizar_imagen(historial: list):
    """Analiza imágenes con GPT-4 Vision."""
    stream = client.chat.completions.create(
        model="gpt-4o",  # ← soporta visión
        messages=[
            {"role": "system",
             "content": "Eres un perito en "
             "bienes raíces. Analiza las fotos "
             "de inmuebles y da estimaciones "
             "de valor, estado y mejoras "
             "posibles. Sé específico y "
             "profesional."},
            *historial
        ],
        max_tokens=800,
        stream=True
    )
    for chunk in stream:
        delta = chunk.choices[0].delta
        if delta.content:
            yield delta.content

📸 GPT-4o puede analizar imágenes y responder preguntas sobre ellas. Casos de uso inmobiliarios: estimar valor, detectar daños, sugerir remodelaciones, verificar el estado del inmueble.

💡 st.file_uploader puede estar dentro o fuera de un st.form. Para uploads + texto juntos, fuera del form es más flexible.

El chat multimodal abre casos de uso muy potentes para bienes raíces. El usuario puede subir una foto de un inmueble y preguntar “¿cuánto crees que vale?” o “¿qué reparaciones necesita?”. GPT-4o analiza la imagen y responde con contexto del mercado. La clave técnica es convertir la imagen a base64 y enviarla como data URL en el mensaje.

Truncamiento inteligente del historial

import streamlit as st
import tiktoken  # pip install tiktoken

# ── Contar tokens (OpenAI) ────────────────
def contar_tokens(historial: list,
                  modelo: str = "gpt-4o-mini"
                  ) -> int:
    enc = tiktoken.encoding_for_model(modelo)
    total = 0
    for msg in historial:
        contenido = msg.get("content", "")
        if isinstance(contenido, str):
            total += len(enc.encode(contenido))
        total += 4  # overhead por mensaje
    return total

# ── Truncar si supera el límite ───────────
MAX_TOKENS = 3000  # dejar margen para respuesta

def gestionar_contexto(historial: list) -> list:
    """Mantiene el system prompt + mensajes
    más recientes dentro del límite de tokens."""
    if not historial:
        return historial

    system = [m for m in historial
              if m["role"] == "system"]
    mensajes = [m for m in historial
                if m["role"] != "system"]

    # Ir eliminando mensajes viejos hasta
    # que quepa en el límite
    while (contar_tokens(system + mensajes)
           > MAX_TOKENS and len(mensajes) > 2):
        mensajes.pop(0)  # eliminar el más viejo

    return system + mensajes

# ── Uso en el chat ────────────────────────
if prompt := st.chat_input("Pregunta..."):
    st.session_state.historial.append(
        {"role": "user", "content": prompt}
    )
    # Antes de llamar al LLM, gestionar contexto
    contexto = gestionar_contexto(
        st.session_state.historial
    )
    respuesta = llamar_llm(contexto)
    # La respuesta va al historial completo
    st.session_state.historial.append(
        {"role": "assistant",
         "content": respuesta}
    )

💰 Por qué importa: GPT-4o cobra ~$5 por millón de tokens de entrada. Una conversación de 100 mensajes sin truncar puede costar 10x más que con truncamiento.

1

System prompt siempre se conserva — es la identidad del asistente

2

Mensajes más recientes tienen prioridad — el usuario necesita que el bot recuerde lo último

✗

Mensajes viejos se eliminan — pérdida de contexto lejano

🧪 Alternativas avanzadas: Resumir mensajes viejos en vez de eliminarlos (summarization), o usar embeddings + vector DB para recuperar contexto relevante (RAG).

El truncamiento de contexto es un problema real en producción. Los modelos tienen un límite de tokens (contexto), y si lo superas, la API devuelve un error. La estrategia más simple es eliminar mensajes viejos. La más avanzada es usar RAG — Retrieval Augmented Generation — donde no almacenas toda la conversación sino que buscas fragmentos relevantes con embeddings.

App Completa — Plataforma Inmobiliaria IA 🏡

Arquitectura de la app final

🔐

Login

stauth

→

🗂️

st.navigation

páginas

→

📋

st.form

filtros

→

⚡

@st.fragment

KPIs live

→

💬

Chat IA

GPT-4o

estructura del proyecto

mi_app_inmobiliaria/
├── .streamlit/
│   └── secrets.toml          # 🔒 gitignore
├── pages/
│   ├── 1_🏠_Buscar.py        # formulario
│   ├── 2_📊_Dashboard.py     # fragmentos
│   └── 3_💬_Asistente.py     # chat
├── utils/
│   ├── auth.py               # autenticación
│   ├── database.py           # consultas DB
│   └── llm.py                # integración IA
├── data/
│   └── propiedades.csv
└── app.py                    # entrada principal

# app.py — punto de entrada
import streamlit as st
import streamlit_authenticator as stauth
from utils.auth import cargar_config_auth

# 1. Configurar autenticación global
config = cargar_config_auth()
auth   = stauth.Authenticate(
    config["credentials"],
    config["cookie"]["name"],
    config["cookie"]["key"],
    config["cookie"]["expiry_days"]
)
auth.login()

if not st.session_state["authentication_status"]:
    st.stop()

# 2. Usuario autenticado — configurar navegación
auth.logout("Cerrar sesión", "sidebar")
st.sidebar.success(
    f"👋 {st.session_state['name']}"
)

# 3. Las páginas se cargan automáticamente
# desde la carpeta /pages

# pages/1_🏠_Buscar.py
import streamlit as st
import pandas as pd

@st.cache_data
def cargar_datos() -> pd.DataFrame:
    return pd.read_csv("data/propiedades.csv")

st.title("🏠 Buscador de Propiedades")

# ── Formulario de búsqueda ────────────────
with st.form("busqueda"):
    col1, col2 = st.columns(2)
    ciudad = col1.selectbox(
        "Ciudad",
        ["Quito", "Guayaquil", "Cuenca"]
    )
    precio_max = col2.slider(
        "Precio máximo", 50_000, 500_000
    )
    min_hab = st.number_input("Hab. mín.", 1, 10)
    buscar  = st.form_submit_button("🔍 Buscar")

if buscar:
    df = cargar_datos()
    resultado = df[
        (df.ciudad == ciudad) &
        (df.precio <= precio_max) &
        (df.habitaciones >= min_hab)
    ]
    st.metric("Resultados", len(resultado))
    st.dataframe(
        resultado,
        column_config={
            "precio": st.column_config.NumberColumn(
                "Precio", format="$%d"
            ),
            "foto": st.column_config.ImageColumn(
                "Foto"
            )
        },
        use_container_width=True
    )

    # ── Panel de acciones (fragment) ──────
    @st.fragment
    def acciones_resultado():
        sel = st.selectbox(
            "Seleccionar propiedad",
            resultado["id"].tolist()
        )
        col_a, col_b, col_c = st.columns(3)
        if col_a.button("⭐ Favorito"):
            st.toast("¡Agregado a favoritos!")
        if col_b.button("📞 Contactar"):
            st.info("Enviando solicitud...")
        if col_c.button("📊 Comparar"):
            st.info("Agregado a comparador")

    acciones_resultado()

Esta es la arquitectura de una app real completa. La carpeta /pages crea automáticamente la navegación lateral en Streamlit. El app.py es el guardian global — si no estás autenticado, ninguna página se carga. Cada página es independiente pero comparte el session_state global incluyendo el rol del usuario.

Dashboard con fragmentos y métricas en vivo

# pages/2_📊_Dashboard.py
import streamlit as st
import plotly.express as px

st.title("📊 Dashboard Inmobiliario")

# ── Filtro de período — NO en form ────────
periodo = st.radio(
    "Período",
    ["7 días", "30 días", "90 días"],
    horizontal=True
)

# ── Sección pesada — carga una vez ────────
@st.cache_data
def cargar_metricas_historicas(periodo):
    # simula consulta a DB
    import pandas as pd, numpy as np
    n = int(periodo.split()[0])
    fechas = pd.date_range(
        end="today", periods=n
    )
    return pd.DataFrame({
        "fecha":  fechas,
        "ventas": np.random.randint(5,25,n),
        "leads":  np.random.randint(20,80,n)
    })

df_hist = cargar_metricas_historicas(periodo)

fig = px.line(
    df_hist, x="fecha",
    y=["ventas", "leads"],
    template="plotly_dark",
    color_discrete_map={
        "ventas": "#FF4B4B",
        "leads":  "#3b82f6"
    }
)
fig.update_layout(
    plot_bgcolor="#0d0d0d",
    paper_bgcolor="#0d0d0d",
    margin=dict(l=0,r=0,t=20,b=0)
)
st.plotly_chart(fig, use_container_width=True)

# ── KPIs en vivo (se actualizan solos) ────
@st.fragment(run_every="15s")
def kpis_tiempo_real():
    import random
    st.divider()
    st.caption("⚡ KPIs en tiempo real")
    m1,m2,m3,m4 = st.columns(4)
    m1.metric("Activos ahora",
              random.randint(10,40),
              random.randint(-3,5))
    m2.metric("Consultas/hora",
              random.randint(8,30),
              random.randint(-2,4))
    m3.metric("Tasa conversión",
              f"{random.uniform(2,8):.1f}%",
              f"{random.uniform(-1,2):.1f}%")
    m4.metric("Precio prom. $k",
              random.randint(85,140),
              random.randint(-5,10))

kpis_tiempo_real()

📊 Evolución de ventas y leads

■ Ventas ■ Leads

⚡ KPIs en tiempo real (cada 15s)

23

Activos

18

Consul/h

4.7%

Conversión

$112k

Prom.

Este es el dashboard completo combinando todas las técnicas del módulo: @st.cache_data para el gráfico histórico, @st.fragment con run_every para los KPIs en tiempo real, y separación clara entre secciones que cambian rápido y secciones que cambian lento. El selectbox de período está fuera del form porque queremos que el gráfico histórico se actualice inmediatamente al cambiar el período.

Manejo de errores y experiencia de usuario

import streamlit as st
import requests
from contextlib import contextmanager

# ── Notificaciones no intrusivas ──────────
def notificar(mensaje: str,
              tipo: str = "info",
              icono: str = ""):
    """Muestra un toast en vez de interrumpir."""
    if tipo == "exito":
        st.toast(f"✅ {mensaje}", icon="✅")
    elif tipo == "error":
        st.toast(f"❌ {mensaje}", icon="❌")
    elif tipo == "advertencia":
        st.toast(f"⚠️ {mensaje}", icon="⚠️")
    else:
        st.toast(mensaje)

# ── Context manager para errores de API ───
@contextmanager
def manejar_errores_api(operacion: str):
    """Envuelve llamadas a APIs externas."""
    try:
        yield
    except requests.exceptions.Timeout:
        st.error(
            f"⏱️ Timeout al {operacion}. "
            f"Intenta de nuevo."
        )
        st.stop()
    except requests.exceptions.ConnectionError:
        st.error(
            "🌐 Sin conexión al servidor. "
            "Verifica tu red."
        )
        st.stop()
    except requests.exceptions.HTTPError as e:
        if e.response.status_code == 429:
            st.warning(
                "⚡ Demasiadas peticiones. "
                "Espera unos segundos."
            )
        elif e.response.status_code >= 500:
            st.error(
                "🔧 Error del servidor. "
                "El equipo fue notificado."
            )
        st.stop()

# ── Uso en el chat ────────────────────────
if prompt := st.chat_input("Pregunta..."):
    with manejar_errores_api("consultar IA"):
        with st.spinner("Consultando IA..."):
            respuesta = llamar_llm(
                st.session_state.historial
            )
    st.session_state.historial.append(
        {"role": "assistant",
         "content": respuesta}
    )
    notificar("Respuesta recibida", "exito")

# ── st.status para operaciones largas ─────
with st.status("Procesando consulta...",
               expanded=True) as status:
    st.write("🔍 Buscando en base de datos...")
    resultados_db = consultar_db(filtros)

    st.write("🤖 Analizando con IA...")
    analisis = analizar_con_llm(resultados_db)

    st.write("📊 Generando visualizaciones...")
    graficos = crear_graficos(resultados_db)

    status.update(
        label="✅ Consulta completada",
        state="complete",
        expanded=False
    )

st.plotly_chart(graficos)
st.markdown(analisis)

# ── Feedback del usuario ───────────────────
col_ok, col_mal = st.columns(2)
if col_ok.button("👍 Útil"):
    notificar("¡Gracias por tu feedback!", "exito")
    registrar_feedback("positivo", analisis)

if col_mal.button("👎 Mejorar"):
    with st.expander("¿Qué mejorarías?"):
        mejora = st.text_area("Cuéntanos...")
        if st.button("Enviar", key="fb_send"):
            registrar_feedback("negativo",
                               analisis, mejora)
            notificar("Feedback registrado","exito")
            st.rerun()

✅ st.toast() muestra notificaciones en la esquina sin interrumpir el flujo. st.status() muestra el progreso de operaciones multi-paso.

El manejo de errores es lo que distingue una app de producción de un prototipo. st.toast es ideal para confirmaciones no intrusivas — aparece en la esquina y desaparece sola. st.status muestra el progreso de operaciones largas de forma visual y profesional. El context manager centraliza el manejo de errores de red para no repetirlo en cada llamada.

Deploy y configuración de producción

# .streamlit/config.toml
# (este SÍ va al repo — no tiene secretos)

[theme]
primaryColor = "#FF4B4B"
backgroundColor = "#0d0d0d"
secondaryBackgroundColor = "#1a1a1a"
textColor = "#e8e8e8"
font = "monospace"

[server]
headless = true
port = 8501
enableCORS = false
enableXsrfProtection = true  # seguridad

[browser]
gatherUsageStats = false  # privacidad

[runner]
fastReruns = true  # mejor rendimiento

📁 config.toml configura el tema y el servidor. secrets.toml guarda credenciales. Los dos van en .streamlit/ pero solo secrets.toml va en .gitignore.

1

Streamlit Cloud (gratis): conecta el repo de GitHub → configura Secrets → deploy automático

2

Docker: streamlit run app.py en un Dockerfile + variables de entorno

3

Hugging Face Spaces: gratis para apps públicas, soporta GPU

4

AWS / GCP / Azure: EC2, Cloud Run, App Service — máximo control

# requirements.txt para deployment
streamlit>=1.33.0
streamlit-authenticator>=0.3.3
openai>=1.0.0
pandas>=2.0.0
plotly>=5.0.0
requests>=2.31.0
python-dotenv>=1.0.0
bcrypt>=4.0.0
PyJWT>=2.8.0
tiktoken>=0.7.0
# FastAPI (si tienes backend separado):
fastapi>=0.111.0
uvicorn>=0.29.0

El config.toml permite personalizar el tema de la app para que coincida con la identidad visual del cliente. Streamlit Cloud es el punto de entrada más fácil para deploy — conectas el repo y en 2 minutos tienes la app en la nube. Para producción seria, Docker + un servidor propio o un servicio cloud da más control y predictibilidad de costos.

Resumen y Reto 🏁

Todo lo que aprendiste hoy

🔒

02

st.secrets

Credenciales fuera del código. secrets.toml en .gitignore. Mismo código en local y en la nube.

🛑

03

Login manual + st.stop()

session_state recuerda el estado entre reruns · st.stop() bloquea el resto del script.

⚡

04

FastAPI + JWT

Backend de auth real. bcrypt para contraseñas · tokens con expiración · arquitectura escalable.

🏆

05

streamlit-authenticator

Multi-usuario listo para producción. Login · logout · registro · recuperación. pip install streamlit-authenticator.

📋

01

st.form

Agrupa widgets y evita re-ejecuciones explosivas. El carrito de compras: ajusta todo y paga una sola vez.

🧩

06

@st.fragment

Re-ejecución parcial. Las secciones pesadas no vuelven a cargar al interactuar con widgets ligeros.

💬

07

Interfaces de Chat

st.chat_message · st.chat_input · st.write_stream · historial persistente en session_state.

🔗

08

Todo integrado

Login seguro → formulario de filtros → fragmento aislado → chat con IA. Una app profesional completa.

Pausa para preguntas. ¿Qué parte les sorprendió más? Generalmente es @st.fragment o el streaming del chat. Dejar tiempo antes del reto.

Reto para casa 🚀

Construir una Plataforma de Consulta Inteligente integrando todo el módulo:

🔐

1. Login seguro

Implementar con streamlit-authenticator
Contraseñas hasheadas con bcrypt
Credenciales en secrets.toml
Bonus: 2 roles (admin / viewer)

📋

2. Formulario de búsqueda

st.form con mínimo 4 filtros
Ciudad, presupuesto, habitaciones, tipo
Resultado filtrando un DataFrame real
Bonus: exportar resultados a CSV

⚡

3. Fragment aislado

@st.fragment para panel de acciones
Guardar favoritos sin recargar la tabla
Métricas que se actualizan independientes
Bonus: run_every=10 para live data

💬

4. Chat con IA

st.chat_input con historial persistente
Efecto streaming con st.write_stream
El chat NO borra los filtros del form
Bonus: conectar con OpenAI API real

🎯 El desafío real: que el formulario de filtros NO se borre cuando el usuario escribe en el chat. Tip: usa @st.fragment en el bloque del chat, o guarda los valores del form en session_state antes de que el chat dispare reruns.

El reto integrador es lo que distingue este módulo. El desafío técnico real es que los filtros del formulario y el historial del chat sobrevivan a los reruns mutuamente. Hay varias soluciones válidas: fragment, session_state para los valores de los filtros, o incluso dividir en páginas con st.navigation. Que experimenten y compartan su solución.

¡A construir! 🚀

Módulo 4 — Completado

Preguntas · Dudas · Reto 🙋

streamlit-authenticator.readthedocs.io

Cierre del módulo. Recursos clave: documentación de streamlit-authenticator, referencia de st.fragment en docs.streamlit.io, y el repositorio de ejemplos del curso. El reto es individual pero se puede presentar en la próxima clase.