Cuando los datos no vienen en un archivo

Hasta aquí los datos llegaban en un archivo. En un trabajo eso pasa poco: lo normal es que estén en una base de datos o detrás de una API 🔌

Una pregunta antes de empezar: ¿de dónde salen los archivos que alguien te manda por correo? Casi siempre de una de estas dos cosas, y alguien los exportó a mano. Este capítulo es para saltarte a esa persona 🙂

Los datos llegan por tres caminos: un archivo con read_csv, una base con sqlite3 donde conviene filtrar y agrupar antes de traer, y una API con requests, siempre con timeout y mirando el código de estado. — El libro entero fue por el primer camino. Los otros dos son los que te vas a encontrar en un trabajo.

Una base de datos, en cuatro líneas

Vamos con tienda.db, la misma base del libro de SQL desde cero. Descárgala y ponla al lado de tu cuaderno.

import sqlite3
import pandas as pd

con = sqlite3.connect('tienda.db')
ticket = pd.read_sql_query('''
    SELECT c.ciudad, COUNT(*) AS pedidos, ROUND(AVG(p.monto), 2) AS ticket
    FROM pedidos p
    JOIN clientes c ON c.id = p.id_cliente
    GROUP BY c.ciudad
    ORDER BY ticket DESC
''', con)
print(ticket.to_string(index=False))

  ciudad  pedidos  ticket
   Piura      167  657.46
Arequipa      168  637.94
    Lima      104  631.17
   Cusco      112  590.60
Trujillo      113  572.77
Chiclayo      212  568.53

Eso es todo: te conectas, escribes SQL y te devuelve un DataFrame 🎉

Y fíjate en el reparto del trabajo, que es la decisión importante: agrupar lo hizo la base, no pandas. Con seiscientas filas da igual, con seis millones no: traértelas todas para agrupar en tu portátil es mover seis millones de filas por la red para quedarte con seis 🚚

La regla que uso: filtra y agrupa en la base, y trae a Python lo que ya está resumido. Con lo que sabes del libro de SQL te alcanza de sobra.

Cuando el filtro viene de fuera

Si la ciudad la elige quien consulta, hay una forma de escribirlo y una sola:

CIUDAD = 'Piura'
sub = pd.read_sql_query(
    'SELECT COUNT(*) AS pedidos FROM pedidos p '
    'JOIN clientes c ON c.id = p.id_cliente WHERE c.ciudad = ?',
    con, params=(CIUDAD,))
print(f'pedidos de {CIUDAD}:', int(sub['pedidos'][0]))

pedidos de Piura: 167

El ? con params aparte, nunca una f-string dentro del SQL. Eso es inyección SQL, tiene capítulo propio en el libro de SQL y es el error de base de datos que más caro sale 🔒

Y ahora una API

Una API es una dirección web que devuelve datos en vez de una página. Le pides algo, te contesta con JSON, y eso en Python es un diccionario.

Para que este capítulo corra en tu máquina aunque no tengas internet, vamos a montar la API aquí mismo. Son quince líneas y no hace falta que las entiendas todas: lo que importa viene después.

import json
import threading
from http.server import BaseHTTPRequestHandler, HTTPServer

PEDIDOS = [{'id': 1, 'ciudad': 'Lima', 'monto': 892.06},
           {'id': 2, 'ciudad': 'Arequipa', 'monto': 731.09},
           {'id': 3, 'ciudad': 'Cusco', 'monto': 407.39}]

class ApiDeMentira(BaseHTTPRequestHandler):
    def do_GET(self):
        if self.path.startswith('/pedidos'):
            cuerpo, codigo, tipo = json.dumps({'total': 3, 'datos': PEDIDOS}), 200, 'application/json'
        elif self.path.startswith('/mantenimiento'):
            cuerpo, codigo, tipo = '<html>Volvemos en un rato</html>', 200, 'text/html'
        else:
            cuerpo, codigo, tipo = json.dumps({'error': 'no existe'}), 404, 'application/json'
        cuerpo = cuerpo.encode()
        self.send_response(codigo)
        self.send_header('Content-Type', tipo)
        self.send_header('Content-Length', str(len(cuerpo)))
        self.end_headers()
        self.wfile.write(cuerpo)
    def log_message(self, *args):
        pass

servidor = HTTPServer(('127.0.0.1', 0), ApiDeMentira)
threading.Thread(target=servidor.serve_forever, daemon=True).start()
BASE = f'http://127.0.0.1:{servidor.server_address[1]}'
print('API de mentira levantada')

API de mentira levantada

Esto que acabamos de hacer tiene nombre y se usa en serio: montar un servidor de mentira para probar código que llama a una API. Si tus pruebas dependen de que internet funcione y de que el proveedor no cambie nada, no son pruebas 🧪

Pedirle datos, que es una línea

import requests

r = requests.get(f'{BASE}/pedidos', timeout=5)
print('codigo :', r.status_code)
print('tipo   :', r.headers['Content-Type'])
datos = r.json()
print('total  :', datos['total'])
print('primero:', datos['datos'][0])

codigo : 200
tipo   : application/json
total  : 3
primero: {'id': 1, 'ciudad': 'Lima', 'monto': 892.06}

Contra una API de verdad se escribe exactamente igual, cambiando BASE por su dirección 🌐

Tres cosas de esa línea, y las tres son costumbres que se pagan:

⏱️ El timeout no es opcional. Sin él, si el otro lado no contesta, tu programa se queda esperando para siempre. Es el fallo más tonto y el que más veces he visto colgar un proceso de madrugada.
🔢 El status_code dice si salió bien. 200 es bien, 404 no existe, 401 no tienes permiso, 429 estás pidiendo demasiado rápido y 500 se rompió el otro lado.
📦 .json() convierte el cuerpo en diccionario. No hace falta json.loads.

Lo que devuelve cuando sale mal

mala = requests.get(f'{BASE}/no-existe', timeout=5)
print('codigo:', mala.status_code, ' ok:', mala.ok)
print('y aun asi trae cuerpo:', mala.json())

codigo: 404  ok: False
y aun asi trae cuerpo: {'error': 'no existe'}

Aquí está la trampa que hay que ver: una respuesta con error no lanza una excepción 😳

requests te devuelve el objeto tan tranquilo, con su 404 y su cuerpo. Si tú no miras el código, tu programa sigue como si nada y guarda {'error': 'no existe'} creyendo que son datos.

Una API que responde no es una API que funcionó. Hay que mirar el código de estado, siempre.

La forma corta de no olvidarse es r.raise_for_status(), que convierte cualquier código de error en una excepción y te obliga a tratarla.

De JSON a tabla

df = pd.DataFrame(datos['datos'])
print(df.to_string(index=False))
print('tipos:')
print(df.dtypes.to_string())

 id   ciudad  monto
  1     Lima 892.06
  2 Arequipa 731.09
  3    Cusco 407.39
tipos:
id          int64
ciudad        str
monto     float64

Una línea y ya estás en el capítulo 12 🐼

Eso funciona porque el JSON venía plano: una lista de diccionarios con las mismas claves. Cuando venga anidado, con diccionarios dentro de diccionarios, está pd.json_normalize, que los aplana en columnas con puntos.

Y mira los tipos, que es la comprobación de siempre: si monto hubiera llegado como texto, ahí se vería, igual que en el capítulo 10 👀

Los tres errores que hay que saber leer

try:
    requests.get('http://127.0.0.1:1/pedidos', timeout=0.3)
except Exception as e:
    print('1. no se pudo ni conectar ->', type(e).__name__)

try:
    requests.get(f'{BASE}/no-existe', timeout=5).raise_for_status()
except Exception as e:
    print('2. contesto con error     ->', type(e).__name__)

try:
    requests.get(f'{BASE}/mantenimiento', timeout=5).json()
except Exception as e:
    print('3. contesto algo que no es JSON ->', type(e).__name__)

1. no se pudo ni conectar -> ConnectionError
2. contesto con error     -> HTTPError
3. contesto algo que no es JSON -> JSONDecodeError

Son tres cosas distintas y conviene no confundirlas 🩺

El primero es de red: no llegaste. El segundo llegó y el otro lado dijo que no. Y el tercero es el más traicionero: contestó 200, todo parecía bien, y lo que mandó fue una página de mantenimiento en HTML.

Ese tercero, sin el try, es así:

requests.get(f'{BASE}/mantenimiento', timeout=5).json()

JSONDecodeError: Expecting value: line 1 column 1 (char 0)

Expecting value: line 1 column 1 quiere decir "lo primero que me diste ya no era JSON". Cuando veas eso, imprime r.text[:200] antes de nada: casi siempre es una página de error o de inicio de sesión 🔍

Lo que falta y no cabe aquí

Si te toca	Qué buscar
La API pide una clave	Va en las cabeceras: `headers={'Authorization': f'Bearer {clave}'}`. Y la clave NUNCA en el código, se lee del entorno
Vienen 10.000 registros de a 100	Paginación. Un bucle que pide páginas hasta que la respuesta viene vacía
Te devuelve 429	Estás pidiendo muy rápido. Se espera y se reintenta, doblando la espera cada vez
Muchas peticiones seguidas	`requests.Session()`, que reutiliza la conexión y va bastante más rápido
Escribir en vez de leer	`requests.post`, igual pero con `json=` para mandar el cuerpo

Y la costumbre que más disgustos ahorra: guarda la respuesta cruda antes de tocarla. Si el proveedor cambia algo mañana, tienes lo de hoy para comparar; y si no, hay que volver a pedirlo todo 💾

La trampa

Un proceso trae los pedidos de una API cada noche y los guarda. Lleva meses funcionando. Una noche el proveedor se cae y a la mañana siguiente el informe sale en cero, sin ningún error en el registro.

import requests, pandas as pd

r = requests.get('https://api.proveedor.com/pedidos')
datos = r.json().get('datos', [])

df = pd.DataFrame(datos)
df.to_csv('pedidos.csv', index=False)
print(f'guardados {len(df)} pedidos')

Qué está mal

Tres cosas y ninguna da error. La primera: no hay timeout, así que si el proveedor tarda en contestar el proceso se queda colgado hasta que alguien lo mate. La segunda: no se mira el status_code, así que un 500 o un 401 pasan de largo. Y la tercera es la que causó el cero: ese .get('datos', []) parece prudente y es justo lo que esconde el problema, porque cuando la respuesta no trae 'datos' devuelve una lista vacía sin quejarse. Después el to_csv pisa el archivo bueno del día anterior con uno vacío, así que además de no traer nada, se perdió lo que ya había. Se arregla con timeout, con raise_for_status(), y sobre todo escribiendo en un archivo temporal y renombrándolo solo si llegaron filas. Guardar nada encima de algo es la parte cara 🚩

Ejercicios

Seis. El 5 es el que de verdad se parece a un encargo 💛

1. Deja que la base haga el trabajo

Trae los cinco clientes que más compraron, con una sola consulta, sin agrupar en pandas.

top = pd.read_sql_query('''
    SELECT c.nombre, ROUND(SUM(p.monto), 2) AS total
    FROM pedidos p JOIN clientes c ON c.id = p.id_cliente
    GROUP BY c.nombre ORDER BY total DESC LIMIT 5
''', con)
print(top.to_string(index=False))

Compara con hacerlo trayendo las 900 filas y agrupando con groupby. El resultado es el mismo y lo que viaja por la red no.

2. Mira lo que llegó antes de creerlo

Antes de .json(), imprime los primeros caracteres de la respuesta.

r = requests.get(f'{BASE}/mantenimiento', timeout=5)
print('codigo:', r.status_code)
print('tipo  :', r.headers['Content-Type'])
print('texto :', r.text[:60])

Código 200 y contenido HTML. Esta es la comprobación de tres líneas que convierte "no entiendo por qué falla" en "ah, está en mantenimiento".

3. Ponle una clave

Añade una cabecera de autorización a la petición y comprueba que la API de mentira la recibe.

r = requests.get(f'{BASE}/pedidos', timeout=5,
                 headers={'Authorization': 'Bearer una-clave-de-mentira'})
print(r.status_code)

La API de mentira no la mira, y da igual: lo que se practica es dónde va. Y la regla que no se negocia: la clave se lee de una variable de entorno con os.environ, nunca escrita en el archivo, porque el archivo termina en algún sitio compartido.

4. Paginar

Escribe el bucle que pide páginas hasta que no vengan más. La API de mentira devuelve siempre lo mismo, así que ponle un tope.

todos = []
for pagina in range(1, 4):
    r = requests.get(f'{BASE}/pedidos?page={pagina}', timeout=5)
    r.raise_for_status()
    lote = r.json()['datos']
    if not lote:
        break
    todos.extend(lote)
print('trajimos', len(todos), 'registros en', pagina, 'paginas')

Fíjate en el break: la condición de salida es que llegue vacío, no un número que te inventes. Un bucle de paginación sin salida clara es la forma más rápida de que te bloqueen la clave.

5. El proceso de la noche, bien hecho

Arregla la trampa del capítulo: timeout, comprobación del código, y que no pise el archivo bueno si no llegó nada.

import os

r = requests.get(f'{BASE}/pedidos', timeout=10)
r.raise_for_status()
datos = r.json()['datos']
if not datos:
    raise ValueError('la API contesto bien pero no trajo filas')

pd.DataFrame(datos).to_csv('pedidos.tmp', index=False)
os.replace('pedidos.tmp', 'pedidos.csv')
print('guardados', len(datos), 'pedidos')

El os.replace del final es la pieza clave: cambia el nombre de golpe, así que o está el archivo viejo entero o el nuevo entero, nunca uno a medias. Es la misma idea que la transacción del libro de SQL.

6. Una API de verdad

Sin dataset y con internet. Busca una API pública que no pida clave y tráete algo a un DataFrame.

Sirve cualquiera: el tipo de cambio, los feriados de Perú, el clima. Lo que vas a descubrir sola es que lo difícil no es pedirlo: es que el JSON viene anidado de una forma que nadie te avisó, y ahí es donde se va la tarde 🧩

Cuando pase, imprime datos.keys() y ve bajando nivel por nivel. Es lo mismo que mirar antes de tocar del capítulo 16.

Comprueba que lo tienes

Tu proceso trae datos de una API cada noche. ¿Cuál de estas cuatro es la que no puede faltar?

El timeout, el raise_for_status y no pisar el archivo si viene vacío
Guardar la respuesta cruda por si acaso
Reintentar tres veces si falla
Avisar por correo cuando termine

Lo que te llevas

🗄️ sqlite3 más pd.read_sql_query y ya tienes la base en un DataFrame.
🚚 Filtra y agrupa en la base. Trae a Python lo que ya está resumido.
🔒 Si el filtro viene de fuera, va con ? y params. Nunca pegado.
⏱️ requests.get siempre con timeout.
🔢 Una respuesta con error no lanza excepción. Hay que mirar el código o llamar a raise_for_status().
🩺 Y los tres errores son distintos: no llegué, llegué y dijo que no, o llegué y me dio algo que no era JSON.

Si lo que quieres es escribir tú las consultas en vez de copiarlas, eso es el libro de SQL desde cero 🗃️

Que tengas lindo día! 🌸