Cómo diseñar scripts reutilizables para consumir múltiples APIs

Introducción

Diseñar scripts para consumir varias APIs requiere reutilización, una configuración explícita y buenos adaptadores. Copiar un programa que funciona, cambiar su URL y repetirlo para cada servicio parece rápido, pero deja varias versiones de la misma lógica que empiezan a divergir en cuanto aparece el primer cambio.

El reto no consiste en escribir una función que acepte cualquier dirección y cualquier credencial. Consiste en compartir las tareas realmente comunes sin borrar las diferencias entre proveedores: autenticación, paginación, formatos, errores y significado de los datos.

En esta guía construiremos una estructura pequeña para clientes HTTP reutilizables. El ejemplo utiliza Python y Requests, limita su alcance a lecturas JSON y separa el transporte de la normalización. Los proveedores y contratos del ejemplo son ficticios; el propósito es aprender el diseño y sus pruebas, no presentar un conector listo para una aplicación comercial concreta.

Índice

• Qué conviene reutilizar y qué debe seguir siendo específico
• Organizar un proyecto pequeño sin crear un marco innecesario
• Configurar destinos y credenciales antes de realizar llamadas
• Ejemplo de cliente reutilizable para lecturas JSON
• Normalizar dos contratos sin confundir sus identidades
• Utilizar el cliente sin incrustar secretos en el código
• Añadir paginación en el adaptador, no mediante suposiciones
• Compartir errores sin automatizar decisiones peligrosas
• Probar reutilización, contratos y fallos sin consumir APIs reales
• Evolucionar el diseño cuando aparezca una necesidad real
• Preguntas frecuentes
• Conclusión

Qué conviene reutilizar y qué debe seguir siendo específico

Reutilizar no significa hacer que todas las APIs parezcan idénticas. Conviene compartir la apertura de conexiones, los límites de espera, la comprobación básica de respuestas y una forma consistente de comunicar fallos. En cambio, los nombres de campos y las reglas que convierten una respuesta en un objeto empresarial pertenecen al adaptador.

Por ejemplo, dos servicios pueden devolver una lista de empresas bajo claves distintas. Esa diferencia es sencilla de adaptar. Pero si uno identifica una organización y el otro identifica una sede, no basta con renombrar campos: hay una diferencia de modelo que debe resolverse antes de unir registros.

Tres responsabilidades, no tres plataformas

El transporte habla HTTP. El adaptador conoce el contrato del proveedor. El proceso de negocio decide qué hacer con los datos normalizados. Puedes mantener estas responsabilidades en módulos de un mismo proyecto; no necesitas desplegar servicios separados para obtener el beneficio.

El proceso debería poder expresar “obtener empresas de esta fuente” sin conocer cómo se construye una cabecera. A su vez, el transporte no debería decidir si una empresa es un cliente activo. Esa decisión necesita información funcional que no pertenece a una llamada de red.

Antes de extraer utilidades, revisa el objetivo del intercambio con el criterio de integrar servicios digitales sin añadir complejidad. La reutilización merece la pena cuando reduce cambios repetidos y permite probar mejor, no cuando solo añade niveles de abstracción.

Organizar un proyecto pequeño sin crear un marco innecesario

La siguiente estructura es una propuesta. Los nombres pueden cambiar; lo importante es que cada archivo tenga una responsabilidad reconocible y que las pruebas puedan ejecutarse sin credenciales reales.

integracion/
  cliente_api.py         # Transporte y errores controlados.
  adaptadores.py         # Contratos y normalización por proveedor.
  proceso.py             # Reglas del trabajo empresarial.
  configuracion.py       # Lectura y validación de parámetros.
  tests/
    test_transporte.py
    test_adaptadores.py
  requisitos.txt         # Dependencias de la combinación probada.
  README.md              # Instalación, ejecución y límites conocidos.

En el ejemplo de esta guía el código se muestra junto para facilitar su lectura. Al crecer, puedes separarlo siguiendo esas responsabilidades. Evita dividir cada función en un archivo distinto si eso hace más difícil seguir el recorrido de una operación.

Establecer un contrato interno pequeño

El transporte devolverá un objeto JSON o un error controlado. El adaptador devolverá una lista de empresas normalizadas. El proceso recibirá esa lista y decidirá si la compara, la guarda o prepara cambios en otro sistema. Ninguna capa imprimirá por su cuenta todos los registros.

Este contrato evita mezclar datos válidos con mensajes de error. Devolver una lista vacía cuando falla la autenticación sería especialmente peligroso: el consumidor podría interpretar que no existen empresas y eliminar información correcta.

Un diseño así también facilita escribir comentarios útiles: documentarás restricciones y decisiones, en lugar de explicar línea a línea lo que el código ya muestra.

Configurar destinos y credenciales antes de realizar llamadas

La configuración debe indicar el origen autorizado, las operaciones disponibles, el entorno y el mecanismo de acceso. No aceptes desde un formulario una URL arbitraria que después recibe una credencial empresarial. Cuanta más libertad tenga el llamador, más comprobaciones necesitará el cliente.

En el diseño propuesto, el llamador utiliza nombres de operación, como companies. Una tabla de configuración los relaciona con rutas predefinidas. Los parámetros de consulta se envían mediante la biblioteca HTTP, no concatenando cadenas manualmente.

Un cliente por proveedor e identidad

Usa instancias separadas para cuentas y servicios diferentes. Así evitas reutilizar accidentalmente cabeceras, cookies o estados de sesión. Las sesiones de Requests permiten persistir ciertos parámetros y reutilizar conexiones; esa comodidad debe acompañarse de una separación explícita de identidades.

El secreto debe proceder del mecanismo autorizado de configuración del entorno. Una variable de entorno puede ser suficiente en un laboratorio controlado, pero no constituye por sí sola un almacén seguro. No la imprimas, no la incorpores al repositorio y revisa quién puede inspeccionar el proceso.

La configuración del ejemplo es de confianza: no debe construirse directamente con datos externos. Sus comprobaciones de URL reducen errores, pero no sustituyen una lista de destinos aprobados y controles de salida de red. Las recomendaciones de OWASP frente a SSRF explican por qué permitir destinos arbitrarios exige defensas adicionales.

Ejemplo de cliente reutilizable para lecturas JSON

El siguiente código está pensado para Python 3.11 o posterior y una instalación de Requests compatible con el entorno. Guárdalo como cliente_api.py. Trabaja con una instancia por ejecución secuencial; no plantea compartir la misma sesión entre varios hilos.

El cliente solo acepta respuestas HTTP 200 con un objeto JSON. Es una decisión deliberada del ejemplo, no una regla para todas las APIs. No sigue redirecciones, no reintenta y limita el cuerpo procesado. Estas restricciones hacen visible el contrato inicial y evitan ocultar decisiones importantes en valores predeterminados.

from __future__ import annotations

import json
from dataclasses import dataclass, field
from typing import Any, Mapping
from urllib.parse import urlsplit

import requests
from requests.adapters import HTTPAdapter


class ApiError(RuntimeError):
    def __init__(self, kind: str, status: int | None = None):
        self.kind = kind
        self.status = status
        super().__init__(kind)  # No URL, token ni cuerpo remoto.


@dataclass(frozen=True)
class ApiConfig:
    origin: str
    routes: Mapping[str, str]
    token: str = field(repr=False)
    max_bytes: int = 1_048_576


class JsonReader:
    """Cliente de lectura JSON; una instancia por proveedor e identidad."""

    def __init__(self, config: ApiConfig):
        u = urlsplit(config.origin)
        if (u.scheme != 'https' or not u.hostname or u.username
                or u.password or u.path not in ('', '/')
                or u.query or u.fragment or u.port not in (None, 443)):
            raise ValueError('El origen debe ser HTTPS sin ruta ni credenciales')
        if (not config.token.strip() or '\r' in config.token
                or '\n' in config.token or config.max_bytes <= 0):
            raise ValueError('Configuración de acceso o tamaño no válida')

        self.urls: dict[str, str] = {}
        for name, path in config.routes.items():
            p = urlsplit(path)
            if (not path.startswith('/') or path.startswith('//')
                    or p.scheme or p.netloc or p.query or p.fragment
                    or '..' in path.split('/') or '\\' in path):
                raise ValueError('Ruta no válida en la configuración')
            self.urls[name] = config.origin.rstrip('/') + path

        self.max_bytes = config.max_bytes
        self.session = requests.Session()
        self.session.trust_env = False  # Sin .netrc ni proxies implícitos.
        self.session.mount('https://', HTTPAdapter(max_retries=0))
        self.session.headers.update({
            'Accept': 'application/json',
            'Authorization': 'Bearer ' + config.token,
            'User-Agent': 'EjemploIntegracion/1.0',
        })

    def __enter__(self) -> JsonReader:
        return self

    def __exit__(self, exc_type, exc_value, traceback) -> None:
        self.session.close()

    def get_object(self, route: str,
                   params: Mapping[str, str | int] | None = None
                   ) -> dict[str, Any]:
        if route not in self.urls:
            raise ValueError('Operación no configurada')
        try:
            with self.session.get(
                self.urls[route], params=params,
                timeout=(3, 15), allow_redirects=False, stream=True,
            ) as response:
                if response.status_code != 200:
                    raise ApiError('unexpected_http_status', response.status_code)
                media = (response.headers.get('Content-Type', '')
                         .split(';')[0].strip().lower())
                if not (media == 'application/json' or media.endswith('+json')):
                    raise ApiError('unexpected_media_type', response.status_code)
                payload = bytearray()
                for chunk in response.iter_content(chunk_size=8192):
                    payload.extend(chunk)
                    if len(payload) > self.max_bytes:
                        raise ApiError('response_too_large', response.status_code)
        except requests.RequestException:
            raise ApiError('transport_error') from None
        finally:
            self.session.cookies.clear()

        def reject_constant(value: str) -> None:
            raise ValueError('Constante no admitida en JSON')

        try:
            data = json.loads(payload.decode('utf-8'), parse_constant=reject_constant)
        except (ValueError, UnicodeDecodeError, RecursionError):
            raise ApiError('invalid_json') from None
        if not isinstance(data, dict):
            raise ApiError('expected_object')
        return data

Qué garantiza y qué no garantiza este ejemplo

El tamaño máximo se aplica a los bytes entregados por iter_content; no debe confundirse con una cuota exacta de bytes de red. El cuerpo se procesa en memoria y el diseño está pensado para respuestas pequeñas. Para grandes exportaciones necesitarías otro tratamiento.

El par timeout=(3, 15) configura esperas de conexión y lectura; no impone un plazo total de dieciocho segundos. La documentación de Requests sobre timeouts advierte que no son un límite global de descarga. Un trabajo que necesite un plazo absoluto requiere control adicional desde su ejecutor.

trust_env=False evita utilizar automáticamente proxies y credenciales de .netrc. También afecta a configuraciones ambientales que Requests podría utilizar. En una red corporativa con proxy o certificados propios, introduce una configuración revisada expresamente; no resuelvas un fallo desactivando la verificación TLS.

Normalizar dos contratos sin confundir sus identidades

Supongamos que el proveedor A devuelve empresas en items, con identificadores de texto en id. El proveedor B utiliza results y account_code. Ambos contratos son ficticios y exigen identificadores de texto; esa exigencia conserva, por ejemplo, un código 0007.

Añade este bloque al archivo anterior. Cada adaptador explica la forma del contrato y produce el mismo tipo interno. La normalización se realiza sin red, por lo que puede probarse con ejemplos pequeños.

@dataclass(frozen=True)
class Company:
    source: str
    external_id: str
    name: str


def normalize_page(data: dict[str, Any], *, source: str,
                   list_key: str, id_key: str, name_key: str
                   ) -> list[Company]:
    rows = data.get(list_key)
    if not isinstance(rows, list):
        raise ApiError('expected_list')
    result: list[Company] = []
    for row in rows:
        if not isinstance(row, dict):
            raise ApiError('expected_record')
        key, name = row.get(id_key), row.get(name_key)
        if not isinstance(key, str) or not key.strip():
            raise ApiError('invalid_identifier')
        if not isinstance(name, str) or not name.strip():
            raise ApiError('invalid_name')
        result.append(Company(source, key, name.strip()))
    return result


def provider_a(data: dict[str, Any]) -> list[Company]:
    return normalize_page(data, source='a', list_key='items',
                          id_key='id', name_key='name')


def provider_b(data: dict[str, Any]) -> list[Company]:
    return normalize_page(data, source='b', list_key='results',
                          id_key='account_code', name_key='display_name')

La identidad interna incluye la fuente. El registro a:0007 no es automáticamente la misma empresa que b:0007. Para relacionarlos necesitas una correspondencia validada o una clave compartida cuyo significado esté acordado. Esta distinción es fundamental para integrar aplicaciones sin duplicar datos.

El ejemplo rechaza una página completa si encuentra un registro inválido. Otra aplicación puede separar válidos y rechazados, pero debe comunicarlo explícitamente. No conviene cambiar a ese comportamiento devolviendo solo los válidos sin informar de lo que falta.

El nombre se limpia de espacios externos; el identificador se conserva tal como llega tras comprobar que no está vacío. No apliques transformaciones agresivas a códigos sin conocer su semántica. Si otro proveedor utiliza identificadores numéricos, crea su adaptador y documenta la conversión en lugar de debilitar todos los contratos.

Utilizar el cliente sin incrustar secretos en el código

Este fragmento muestra cómo se conectan las piezas. El dominio bajo .example es ficticio y no ofrece una API real. Antes de ejecutar una consulta real, sustituye el origen y la ruta por los aprobados y adapta el contrato al servicio correspondiente.

import os
from cliente_api import ApiConfig, ApiError, JsonReader, provider_a

config = ApiConfig(
    origin='https://proveedor-a.example',
    routes={'companies': '/v1/companies'},
    token=os.environ['API_A_TOKEN'],
)

try:
    with JsonReader(config) as client:
        page = client.get_object('companies', {'page_size': 50})
    companies = provider_a(page)
except ApiError as error:
    # Registrar solo clasificación y estado; no imprimir datos remotos.
    print({'error': error.kind, 'status': error.status})
    raise SystemExit(1)
else:
    print({'records_in_page': len(companies)})

El resultado corresponde a una página, no a todo el catálogo. El nombre de la variable y el mensaje lo indican para evitar una interpretación falsa. Si falta la variable de entorno, la configuración falla antes de realizar ninguna petición.

Evitar consumidores que interpreten cualquier salida como éxito

Cuando un script forma parte de un proceso programado, define su señal de terminación y el lugar donde publica resultados. Un error no debería dejar un archivo parcial con el nombre del informe definitivo. Puedes preparar la salida por separado y publicarla únicamente tras superar los controles previstos.

También distingue ejecución sin datos de ejecución fallida. Un conjunto vacío validado puede ser un resultado legítimo. Un fallo de red no aporta ninguna evidencia sobre cuántos registros existen en el servicio.

Añadir paginación en el adaptador, no mediante suposiciones

Las páginas pueden identificarse por números, cursores o enlaces. El cliente común no debería asumir que todas las respuestas incluyen un campo next ni que una página corta significa necesariamente el final. La condición de terminación pertenece al contrato de cada proveedor.

El ejemplo anterior no implementa paginación deliberadamente. Para incorporarla, define cómo se obtiene la primera página, qué indica que hay otra y qué información permite reanudar una descarga interrumpida. Documenta además si el cursor caduca o si representa una instantánea consistente.

Establecer defensas contra recorridos interminables

Propongo limitar páginas por ejecución, detectar cursores repetidos y registrar cuántos elementos se han procesado. Si se alcanza un límite de seguridad antes del final, devuelve un estado de incompleto; no presentes el resultado como catálogo íntegro.

Cuando el servidor devuelve una URL de continuación, trátala como información externa. Valida destino y esquema antes de reenviar credenciales. No pases automáticamente esa URL a una sesión autenticada. Si el proveedor requiere otro host, debe estar contemplado expresamente en el diseño.

También puede haber registros que cambien mientras se recorren las páginas. La estrategia de consistencia depende del servicio: instantáneas, marcas de cambio o conciliaciones posteriores. Para un catálogo empresarial, revisar solo el número de páginas no demuestra que no falten elementos.

La conexión con los pipelines de datos empresariales está en conservar entrada, progreso y salida como etapas distintas, no en ocultarlas detrás de una función que dice devolverlo todo.

Compartir errores sin automatizar decisiones peligrosas

La clase ApiError del ejemplo proporciona una clasificación estable y, cuando existe, un código HTTP. El proceso puede decidir si detiene el trabajo, avisa o lo vuelve a programar. No tiene que analizar frases cambiantes del proveedor para saber que el cuerpo no era JSON.

Esta clasificación podría ampliarse a autenticación, cuota o contrato. Hazlo conservando la distinción entre no haber recibido una respuesta HTTP y haber recibido una respuesta que rechaza la operación. Un código inventado como 0 no debe confundirse con un estado HTTP real.

No añadir escrituras como una variación trivial

Incorporar un parámetro method y permitir cualquier verbo amplía mucho el riesgo del cliente. Crear un pedido necesita criterios sobre duplicidad, resultado desconocido y confirmación que una lectura no requiere en el mismo grado.

La definición HTTP de idempotencia no convierte todas las llamadas en seguras para repetir. Antes de añadir reintentos de escritura, comprueba las garantías de la operación y del proveedor. Una cabecera de idempotencia solo sirve cuando el servicio la admite y aplica.

Por eso el ejemplo no reintenta dentro de la biblioteca. Un ejecutor puede introducir una política acotada para lecturas, pero debe contabilizar cada intento y respetar las instrucciones del proveedor. La política de recuperación no debería aparecer accidentalmente en dos capas y multiplicar las llamadas.

Probar reutilización, contratos y fallos sin consumir APIs reales

Empieza por las funciones que no dependen de la red. Comprueba que dos contratos producen el mismo modelo interno sin borrar la procedencia. También verifica que el programa rechaza los datos que no sabe interpretar.

from cliente_api import ApiError, provider_a, provider_b

left = provider_a({'items': [{'id': '0007', 'name': ' Taller Norte '}]})
right = provider_b({'results': [
    {'account_code': '0007', 'display_name': 'Taller Norte'}
]})
assert left[0].external_id == '0007'
assert left[0].name == 'Taller Norte'
assert left[0].source != right[0].source
assert provider_a({'items': []}) == []

try:
    provider_a({'items': [{'id': None, 'name': 'Taller Norte'}]})
except ApiError as error:
    assert error.kind == 'invalid_identifier'
else:
    raise AssertionError('El adaptador aceptó un identificador inválido')

Simular el transporte

Sustituye la sesión HTTP por una respuesta controlada durante las pruebas, o utiliza una herramienta de simulación compatible con tu proyecto. Verifica HTTP inesperado, tipo de contenido incorrecto, JSON roto, cuerpo excesivo, timeout y ruta no configurada. Comprueba también que la sesión cierra recursos cuando se produce una excepción.

No basta con comprobar resultados. Inspecciona los argumentos de la petición simulada: tiempo de espera, ausencia de redirecciones, parámetros y destino. Una prueba puede demostrar que el cliente nunca envía una credencial al host de otro proveedor.

Comprobar compatibilidad después de modificar lo común

Cuando cambies el cliente, ejecuta las pruebas de todos los adaptadores que lo utilizan. Una mejora para un proveedor no debe introducir una nueva suposición sobre el resto. Conserva como pruebas los errores descubiertos durante integraciones reales, con sus datos desidentificados o sintéticos.

Evolucionar el diseño cuando aparezca una necesidad real

Un cliente reutilizable pequeño puede crecer mediante capacidades concretas: autenticación adicional, carga de ficheros, paginación, reanudación o concurrencia. Añade cada una cuando conozcas su contrato y tengas casos de prueba. Intentar prever todas las APIs posibles suele producir una interfaz difícil de utilizar.

Autenticación extensible

El ejemplo presupone un token Bearer ya obtenido. No implementa OAuth ni renovación de tokens. Un proveedor que exige firma de solicitudes necesita otro componente; no basta con cambiar el nombre de la cabecera. Mantén esa lógica asociada a su identidad y evita exponer el secreto al proceso de negocio.

Concurrencia con límites compartidos

Aumentar trabajadores requiere revisar sesiones, cuotas y orden de operaciones. Tener varios objetos cliente no significa disponer de varias cuotas independientes. Si todos usan la misma cuenta, el control de consumo debe coordinarse en el ámbito correspondiente.

Para un negocio pequeño, una ejecución secuencial con puntos de control puede ser preferible a un conjunto de tareas paralelas sin recuperación clara. El cambio debe responder a una limitación medida, no a la idea de que paralelo siempre es mejor.

Publicar una biblioteca interna

Cuando varias integraciones consuman realmente el mismo módulo, versiona su interfaz y sus cambios. Mantén ejemplos, pruebas y un procedimiento de actualización. Hasta entonces, un módulo compartido dentro de un repositorio puede ser suficiente.

La documentación debería indicar capacidades y límites, con el enfoque de documentación tecnológica sencilla. Un consumidor necesita saber tanto qué puede llamar como qué debe resolver por su cuenta.

Preguntas frecuentes

Conclusión

Un script reutilizable para múltiples APIs combina una base común pequeña con adaptadores que conocen sus contratos. La configuración define destinos e identidades; el transporte controla la comunicación; el adaptador valida y normaliza; el proceso empresarial decide qué hacer.

Empieza con una operación de lectura, prueba sus fallos y añade un segundo proveedor sin modificar las reglas del primero. Esa es una comprobación más valiosa de reutilización que admitir decenas de parámetros que nadie sabe utilizar con seguridad.