Saltar al contenido
🐍 Curso de Python
/

Módulo 5 · Lección 10 de 30

Proyecto: un cliente de una API pública

Cuatro capas, veintitrés pruebas sin tocar internet, y dos sorpresas: un fallo que encontró la suite y un número que dice la verdad y miente.

⏱ 22 min de lectura

Toca juntarlo todo. Vamos a escribir un programa de terminal que consulta la API de GitHub, y lo vamos a escribir como se escribe algo que va a durar: con las capas separadas y con pruebas que no necesitan red.

Salida en pantalla
repos ver django/django
repos incidencias psf/requests -n 6
repos resumen psf/requests -n 100
repos --json ver psf/requests | jq .estrellas

Las cuatro capas#

La decisión de arquitectura es esta, y es la que hace que el programa se pueda probar:

Salida en pantalla
repos/
  errores.py     los fallos posibles, y nada más
  cliente.py     HTTP: la ÚNICA capa que sabe de requests
  modelos.py     lo que el programa usa, separado de lo que manda la API
  servicio.py    las operaciones que tienen sentido
  cli.py         argumentos y formato. No sabe de HTTP
tests/
  servidor_falso.py
  test_cliente.py

La regla que lo sostiene: requests solo se importa en cliente.py. Si mañana hay que cambiar de biblioteca, se toca un fichero. Y más importante: todo lo demás se puede probar sin red.

Los errores propios#

Antes que nada, los fallos que el programa puede tener. En su vocabulario, no en el de HTTP:

python
class ErrorDeRepos(Exception):
        """Cualquier problema del cliente. Todo lo demás hereda de aquí."""


class SinRed(ErrorDeRepos):
        """No hubo respuesta: DNS, conexión rechazada, tiempo agotado."""


class ErrorDeLaApi(ErrorDeRepos):
        """El servidor contestó, y contestó algo que no sirve."""

    def __init__(self, codigo, mensaje, url):
                self.codigo = codigo
        self.mensaje = mensaje
        self.url = url
        super().__init__(f'{codigo} en {url}: {mensaje}')


class NoEncontrado(ErrorDeLaApi): ...
class NoAutorizado(ErrorDeLaApi): ...


class CuotaAgotada(ErrorDeLaApi):
        def __init__(self, codigo, mensaje, url, reinicio=None):
                self.reinicio = reinicio
        super().__init__(codigo, mensaje, url)

    @property
    def segundos_de_espera(self):
                if self.reinicio is None:
                        return None
        return max(0, self.reinicio - time.time())

Tres decisiones:

Una raíz común, para que quien use esto pueda protegerse con un except y punto. Es la misma idea que RequestException, de la lección 5.

SinRed y ErrorDeLaApi separados, que es la distinción de urllib de la lección 4 y la que decide si se reintenta.

CuotaAgotada sabe cuánto esperar. Es el único error que trae una instrucción de qué hacer, y por eso merece su propia clase.

El cliente#

python
class Cliente:
        def __init__(self, token=None, base=BASE, margen_de_cuota=2):
                self.base = base.rstrip('/')
        self.margen_de_cuota = margen_de_cuota
        self.sesion = requests.Session()
        self.sesion.headers.update({
            'Accept': 'application/vnd.github+json',
            'X-GitHub-Api-Version': '2022-11-28',
            'User-Agent': AGENTE,
        })

        token = token or os.environ.get('TOKEN_GITHUB')
        if token:
                        self.sesion.headers['Authorization'] = f'Bearer {token}'

        reintentos = Retry(
            total=3,
            backoff_factor=0.5,
            backoff_jitter=1.0,
            status_forcelist=(429, 500, 502, 503, 504),
            allowed_methods=frozenset({'GET', 'HEAD'}),
            respect_retry_after_header=True,
            raise_on_status=False,
        )
        adaptador = HTTPAdapter(max_retries=reintentos)
        self.sesion.mount('https://', adaptador)
        self.sesion.mount('http://', adaptador)

Todo lo del bloque, junto:

base es un parámetro. Parece un detalle y es lo que permite apuntar al servidor de pruebas. Sin eso, no hay suite.

El 403 no está en la lista de reintentos, a propósito y en contra de lo que parecería. Es la medición de la lección 9: en esta API el 403 es la cuota, no trae cabecera de espera, y el reinicio tarda cuarenta minutos. Reintentar tres veces con espera exponencial son tres golpes a una puerta cerrada.

raise_on_status=False, para que al agotarse los reintentos llegue la respuesta y podamos leer el cuerpo, que es donde la API explica el problema.

Los dos esquemas montados. Volvemos a esto, porque tiene historia.

El recorrido de páginas es el generador de la lección 9:

python
def recorrer(self, ruta, limite=None, **parametros):
        url = self._url(ruta)
    vistas = set()
    entregados = 0

    while url and url not in vistas:
                vistas.add(url)
        respuesta = self._pedir(url, parametros)
        parametros = None          # el enlace siguiente ya los trae dentro

        lote = respuesta.json()
        if not lote:
                        return
        for elemento in lote:
                        yield elemento
            entregados += 1
            if limite is not None and entregados >= limite:
                                return

        self._respetar_cuota(respuesta)
        url = respuesta.links.get('next', {}).get('url')

Y la traducción de respuestas a errores propios:

python
def _traducir(self, respuesta):
        url = respuesta.url.split('?')[0]     # sin parámetros: pueden traer secretos
    mensaje = self._motivo(respuesta)
    codigo = respuesta.status_code

    agotada = respuesta.headers.get('x-ratelimit-remaining') == '0'
    if codigo in (403, 429) and agotada:
                reinicio = respuesta.headers.get('x-ratelimit-reset')
        return CuotaAgotada(codigo, mensaje, url,
                            int(reinicio) if reinicio else None)

    if codigo == 404:
                return NoEncontrado(codigo, mensaje, url)
    if codigo in (401, 403):
                return NoAutorizado(codigo, mensaje, url)
    return ErrorDeLaApi(codigo, mensaje, url)

Fíjate en cómo se distinguen los dos 403: por la cabecera de cuota, no por el código. Con la cuota a cero es el límite; con cuota disponible es que no tienes permiso. Son problemas distintos con soluciones distintas, y la API los manda con el mismo número.

Y el recorte de los parámetros en url es la lección 7: si alguien pasa un secreto por la consulta, no queremos que acabe en el texto de una excepción que irá a un registro.

Los modelos#

Aquí está la idea que más se salta la gente: no pasear diccionarios de la API por todo el programa.

python
@dataclass(frozen=True, slots=True)
class Incidencia:
        numero: int
    titulo: str
    estado: str
    autor: str
    etiquetas: tuple[str, ...]
    creada: datetime.datetime | None
    es_propuesta: bool

    @classmethod
    def desde_api(cls, datos):
                return cls(
            numero=datos['number'],
            titulo=datos.get('title', ''),
            estado=datos.get('state', ''),
            autor=(datos.get('user') or {}).get('login', '(sin autor)'),
            etiquetas=tuple(e['name'] for e in datos.get('labels', [])),
            creada=_fecha(datos.get('created_at')),
            # La API devuelve las propuestas de cambio mezcladas con las
            # incidencias, y solo se distinguen por esta clave.
            es_propuesta='pull_request' in datos,
        )

Lo que se gana:

Los nombres son los tuyos. estrellas y no stargazers_count. El programa se lee en su idioma.

Las fechas ya son fechas, convertidas en un solo sitio con la función de la lección 3.

Los ochenta y seis campos se quedan en ocho. Los que usas.

Hay un solo sitio donde tocar si la API cambia un nombre.

Y el (datos.get('user') or {}) no es paranoia: cuando se borra una cuenta de GitHub, ese campo llega como nulo. Con datos['user']['login'] eso es un TypeError en producción y nunca en pruebas.

El servicio y la interfaz#

El servicio son las operaciones con sentido, sin saber de HTTP:

python
def resumen(self, nombre, limite=200):
        abiertas = list(self.incidencias(nombre, limite=limite))
    return {
        'total': len(abiertas),
        'por_etiqueta': collections.Counter(
            e for i in abiertas for e in i.etiquetas
        ).most_common(10),
        'por_autor': collections.Counter(i.autor for i in abiertas).most_common(10),
        'sin_etiquetar': sum(1 for i in abiertas if not i.etiquetas),
        'mas_antigua': min((i for i in abiertas if i.creada),
                           key=lambda i: i.creada, default=None),
    }

Y la interfaz traduce errores a mensajes y a códigos de salida, que es lo del módulo 2:

python
try:
        with Cliente() as cliente:
                salida = _ejecutar(argumentos, Repos(cliente))
except NoEncontrado as e:
        print(f'No existe: {e.url}', file=sys.stderr)
    return 4
except CuotaAgotada as e:
        espera = e.segundos_de_espera
    print(f'Cuota agotada. Vuelve en {espera / 60:.0f} minutos.', file=sys.stderr)
    if not _hay_token():
                print('Con un token tendrías 5000 peticiones por hora en vez de 60.',
              file=sys.stderr)
    return 3
except ErrorDeRepos as e:
        print(f'Error: {e}', file=sys.stderr)
    return 2
Salida en pantalla
no existe  -> 4
bien       -> 0
sin orden  -> 2

El consejo del token en el mensaje de cuota es lo que distingue un error útil de uno que solo informa: dice qué hacer, y sale solo cuando aplica.

El servidor de mentira#

Las pruebas no tocan internet. Un servidor HTTP de la biblioteca estándar en un hilo, al que se le dice qué contestar:

python
class ServidorFalso:
        def responder(self, ruta, cuerpo, codigo=200, cabeceras=None, veces=None):
                self.respuestas.setdefault(ruta, []).append(
            {'cuerpo': cuerpo, 'codigo': codigo,
             'cabeceras': cabeceras or {}, 'veces': veces}
        )

Se puede encolar más de una respuesta por ruta, que es lo que permite probar los reintentos: dos veces 503 y luego bien. Y apunta lo que le piden, que es lo que permite comprobar las cabeceras enviadas.

Por qué un servidor de verdad en lugar de sustituir funciones: porque prueba la biblioteca de verdad, con sus reintentos, sus conexiones y su análisis de la cabecera de navegación. Un sustituto de requests.get no habría encontrado el fallo que viene ahora.

Las veintitrés pruebas#

Salida en pantalla
Ran 23 tests in 17.312s
OK

Agrupadas en cinco bloques: peticiones y cabeceras, errores, reintentos, paginación y cuota. Las que más valen:

python
def test_403_con_cuota_a_cero_es_cuota_agotada(self):
        reinicio = int(time.time()) + 600
    self.servidor.responder(
        '/x', {'message': 'API rate limit exceeded'}, 403,
        {'X-RateLimit-Remaining': '0', 'X-RateLimit-Reset': str(reinicio)},
    )
    with self.assertRaises(CuotaAgotada) as caso:
                self.cliente().obtener('/x')
    self.assertAlmostEqual(caso.exception.segundos_de_espera, 600, delta=5)


def test_los_parametros_solo_van_en_la_primera(self):
        self.servidor.responder('/l', [{'n': 1}], cabeceras=self._enlazar('/l2?cursor=xyz'))
    self.servidor.responder('/l2', [{'n': 2}])
    list(self.cliente().recorrer('/l', state='open'))
    self.assertIn('state=open', self.servidor.peticiones[0]['ruta'])
    self.assertEqual(self.servidor.peticiones[1]['ruta'], '/l2?cursor=xyz')


def test_es_perezoso(self):
        generador = self.cliente().recorrer('/l')
    self.assertEqual(len(self.servidor.peticiones), 0)   # aún no ha pedido nada
    next(generador)
    self.assertEqual(len(self.servidor.peticiones), 1)


def test_el_error_no_guarda_los_parametros(self):
        with self.assertRaises(NoEncontrado) as caso:
                self.cliente().obtener('/x', token_secreto='NO-DEBE-SALIR')
    self.assertNotIn('NO-DEBE-SALIR', str(caso.exception))

La última es de las que más valen por línea escrita: comprueba que un secreto no acaba en el texto de una excepción. Es una propiedad de seguridad, y sin prueba se pierde en el primer refactor.