Módulo 3: JSON & Tools

API Client Reutilizable: Un Patrón Profesional

Siguiente →

Descripción de la cápsula

Hasta ahora, cada vez que necesitas consumir una API, escribes requests.get() con timeout, headers, error handling... y lo repites en cada llamada. Cuando tienes 5 peticiones a la misma API, el código se llena de duplicación: la misma URL base, los mismos headers, el mismo manejo de errores copiado 5 veces. Eso no escala.

Esta cápsula te enseña el patrón de API Client reutilizable: una clase Python que encapsula toda la lógica HTTP — sesión, headers, autenticación, timeout, error handling — y expone métodos limpios como .get("/users/octocat") que hacen todo el trabajo pesado por ti. Es el salto de "scripts que hacen requests" a "código profesional que consume APIs".

El API Client que construyas aquí es la base directa del REST Client CLI del Módulo 4. En ese proyecto vas a crear una instancia de APIClient para GitHub, otra para JSONPlaceholder, otra para cada API que agregues — y cada una funciona igual. Escríbelo una vez, úsalo con cualquier API REST.

Por qué requests.Session importa

En la cápsula 06 de Módulo 1 viste requests.Session() brevemente. Aquí lo vas a entender a fondo, porque es la base del API Client.

Sin Session: repetición constante

import requests

headers = {
    "Accept": "application/vnd.github.v3+json",
    "User-Agent": "MiApp/1.0"
}

r1 = requests.get("https://api.github.com/users/octocat",
                   headers=headers, timeout=10)
r2 = requests.get("https://api.github.com/users/mojombo",
                   headers=headers, timeout=10)
r3 = requests.get("https://api.github.com/users/defunkt",
                   headers=headers, timeout=10)

for r in [r1, r2, r3]:
    print(f"{r.json()['login']:15s} → {r.json()['public_repos']} repos")

Output esperado:

octocat         → 8 repos
mojombo         → 66 repos
defunkt         → 107 repos

Funciona, pero headers=headers, timeout=10 se repite en cada línea. Y cada petición abre una nueva conexión TCP.

Con Session: configuración centralizada

import requests

session = requests.Session()
session.headers.update({
    "Accept": "application/vnd.github.v3+json",
    "User-Agent": "MiApp/1.0"
})

r1 = session.get("https://api.github.com/users/octocat", timeout=10)
r2 = session.get("https://api.github.com/users/mojombo", timeout=10)
r3 = session.get("https://api.github.com/users/defunkt", timeout=10)

for r in [r1, r2, r3]:
    print(f"{r.json()['login']:15s} → {r.json()['public_repos']} repos")

session.close()

Output esperado:

octocat         → 8 repos
mojombo         → 66 repos
defunkt         → 107 repos

Los headers se definen una vez. La sesión reutiliza conexiones TCP (keep-alive) — la segunda y tercera petición al mismo host son más rápidas porque no necesitan un nuevo TCP handshake.

Midiendo la diferencia de performance

import requests
import time

users = ["octocat", "mojombo", "defunkt", "schacon", "pjhyett"]

print("=== Sin Session (nueva conexión cada vez) ===")
start = time.time()
for user in users:
    requests.get(f"https://api.github.com/users/{user}",
                 headers={"Accept": "application/json"}, timeout=10)
no_session_ms = (time.time() - start) * 1000
print(f"  Tiempo total: {no_session_ms:.0f}ms")

print("\n=== Con Session (conexión reutilizada) ===")
session = requests.Session()
session.headers["Accept"] = "application/json"
start = time.time()
for user in users:
    session.get(f"https://api.github.com/users/{user}", timeout=10)
session_ms = (time.time() - start) * 1000
session.close()
print(f"  Tiempo total: {session_ms:.0f}ms")

if no_session_ms > session_ms:
    print(f"\n  Session fue {no_session_ms - session_ms:.0f}ms más rápida")

La diferencia varía según tu red, pero con Session las peticiones subsecuentes al mismo host son consistentemente más rápidas por la reutilización de conexiones.

Construyendo el APIClient paso a paso

Paso 1: Estructura base con Session y base URL

requests.Session no tiene base_url nativo. Lo agregamos como atributo de la clase:

import requests

class APIClient:
    """Cliente HTTP reutilizable con base URL y configuración centralizada."""

    def __init__(self, base_url, timeout=10):
        self.base_url = base_url.rstrip("/")
        self.timeout = timeout
        self.session = requests.Session()
        self.session.headers.update({
            "Accept": "application/json",
            "User-Agent": "PythonAPIClient/1.0"
        })

    def _build_url(self, path):
        """Construye la URL completa a partir de un path relativo."""
        if path.startswith("http"):
            return path
        return f"{self.base_url}/{path.lstrip('/')}"

    def close(self):
        """Cierra la sesión HTTP y libera recursos."""
        self.session.close()


client = APIClient("https://api.github.com")
print(f"Base URL: {client.base_url}")
print(f"URL para /users: {client._build_url('/users/octocat')}")
print(f"URL absoluta: {client._build_url('https://otro.com/api')}")
client.close()

Output esperado:

Base URL: https://api.github.com
URL para /users: https://api.github.com/users/octocat
URL absoluta: https://otro.com/api

rstrip("/") en el constructor y lstrip("/") en _build_url() evitan URLs con doble slash como https://api.github.com//users.

Paso 2: Agregar autenticación

import requests

class APIClient:
    """Cliente HTTP con autenticación configurable."""

    def __init__(self, base_url, timeout=10, token=None, api_key=None,
                 api_key_header="X-API-Key"):
        self.base_url = base_url.rstrip("/")
        self.timeout = timeout
        self.session = requests.Session()
        self.session.headers.update({
            "Accept": "application/json",
            "User-Agent": "PythonAPIClient/1.0"
        })

        if token:
            self.session.headers["Authorization"] = f"Bearer {token}"
        if api_key:
            self.session.headers[api_key_header] = api_key

    def _build_url(self, path):
        if path.startswith("http"):
            return path
        return f"{self.base_url}/{path.lstrip('/')}"

    def close(self):
        self.session.close()


github = APIClient("https://api.github.com")
print(f"Headers sin auth: {dict(github.session.headers)}")

github_auth = APIClient("https://api.github.com", token="ghp_example123")
print(f"\nHeaders con token: {dict(github_auth.session.headers)}")

weather = APIClient("https://api.openweathermap.org", api_key="abc123", api_key_header="X-API-Key")
print(f"\nHeaders con API key: {dict(weather.session.headers)}")

github.close()
github_auth.close()
weather.close()

Output esperado:

Headers sin auth: {'User-Agent': 'PythonAPIClient/1.0', 'Accept-Encoding': 'gzip, deflate', 'Accept': 'application/json', 'Connection': 'keep-alive'}

Headers con token: {'User-Agent': 'PythonAPIClient/1.0', 'Accept-Encoding': 'gzip, deflate', 'Accept': 'application/json', 'Connection': 'keep-alive', 'Authorization': 'Bearer ghp_example123'}

Headers con API key: {'User-Agent': 'PythonAPIClient/1.0', 'Accept-Encoding': 'gzip, deflate', 'Accept': 'application/json', 'Connection': 'keep-alive', 'X-API-Key': 'abc123'}

La autenticación se configura una vez en el constructor y se envía automáticamente en cada petición.

Paso 3: El método _request() con error handling

El corazón del APIClient es un método privado _request() que centraliza toda la lógica HTTP:

import requests
import json

class APIClient:
    """Cliente HTTP con error handling integrado."""

    def __init__(self, base_url, timeout=10, token=None):
        self.base_url = base_url.rstrip("/")
        self.timeout = timeout
        self.session = requests.Session()
        self.session.headers.update({
            "Accept": "application/json",
            "User-Agent": "PythonAPIClient/1.0"
        })
        if token:
            self.session.headers["Authorization"] = f"Bearer {token}"

    def _build_url(self, path):
        if path.startswith("http"):
            return path
        return f"{self.base_url}/{path.lstrip('/')}"

    def _request(self, method, path, **kwargs):
        """
        Ejecuta un request HTTP con error handling completo.
        Retorna dict con: success, status, data, error, error_type
        """
        url = self._build_url(path)
        kwargs.setdefault("timeout", self.timeout)

        try:
            response = self.session.request(method, url, **kwargs)
        except requests.exceptions.ConnectionError:
            return {"success": False, "status": None, "data": None,
                    "error": f"No se pudo conectar a {self.base_url}",
                    "error_type": "CONNECTION"}
        except requests.exceptions.Timeout:
            return {"success": False, "status": None, "data": None,
                    "error": f"Timeout después de {kwargs['timeout']}s",
                    "error_type": "TIMEOUT"}
        except requests.exceptions.RequestException as e:
            return {"success": False, "status": None, "data": None,
                    "error": str(e), "error_type": "NETWORK"}

        if response.status_code == 204:
            return {"success": True, "status": 204, "data": None,
                    "error": None, "error_type": None}

        content_type = response.headers.get("Content-Type", "")
        data = None

        if "application/json" in content_type and response.text:
            try:
                data = response.json()
            except (json.JSONDecodeError, requests.exceptions.JSONDecodeError):
                return {"success": False, "status": response.status_code,
                        "data": None, "error": "JSON malformado en respuesta",
                        "error_type": "JSON_PARSE"}

        if not response.ok:
            error_msg = f"HTTP {response.status_code} {response.reason}"
            if isinstance(data, dict) and "message" in data:
                error_msg = f"{error_msg}: {data['message']}"
            return {"success": False, "status": response.status_code,
                    "data": data, "error": error_msg, "error_type": "HTTP_ERROR"}

        return {"success": True, "status": response.status_code,
                "data": data, "error": None, "error_type": None}

    def close(self):
        self.session.close()


github = APIClient("https://api.github.com")

result = github._request("GET", "/users/octocat")
print(f"GET user: {'✅' if result['success'] else '❌'} {result['status']}")
if result["success"]:
    print(f"  Login: {result['data']['login']}")

result = github._request("GET", "/users/no-existe-xyz-99999")
print(f"\nGET 404:  {'✅' if result['success'] else '❌'} {result['error']}")

github.close()

Output esperado:

GET user: ✅ 200
  Login: octocat

GET 404:  ❌ HTTP 404 Not Found: Not Found

Paso 4: Métodos públicos get(), post(), patch(), delete()

Los métodos públicos son wrappers limpios sobre _request():

import requests
import json

class APIClient:
    """Cliente HTTP reutilizable completo."""

    def __init__(self, base_url, timeout=10, token=None):
        self.base_url = base_url.rstrip("/")
        self.timeout = timeout
        self.session = requests.Session()
        self.session.headers.update({
            "Accept": "application/json",
            "User-Agent": "PythonAPIClient/1.0"
        })
        if token:
            self.session.headers["Authorization"] = f"Bearer {token}"

    def _build_url(self, path):
        if path.startswith("http"):
            return path
        return f"{self.base_url}/{path.lstrip('/')}"

    def _request(self, method, path, **kwargs):
        url = self._build_url(path)
        kwargs.setdefault("timeout", self.timeout)
        try:
            response = self.session.request(method, url, **kwargs)
        except requests.exceptions.ConnectionError:
            return {"success": False, "status": None, "data": None,
                    "error": f"No se pudo conectar a {self.base_url}",
                    "error_type": "CONNECTION"}
        except requests.exceptions.Timeout:
            return {"success": False, "status": None, "data": None,
                    "error": f"Timeout después de {kwargs['timeout']}s",
                    "error_type": "TIMEOUT"}
        except requests.exceptions.RequestException as e:
            return {"success": False, "status": None, "data": None,
                    "error": str(e), "error_type": "NETWORK"}

        if response.status_code == 204:
            return {"success": True, "status": 204, "data": None,
                    "error": None, "error_type": None}

        content_type = response.headers.get("Content-Type", "")
        data = None
        if "application/json" in content_type and response.text:
            try:
                data = response.json()
            except (json.JSONDecodeError, requests.exceptions.JSONDecodeError):
                return {"success": False, "status": response.status_code,
                        "data": None, "error": "JSON malformado",
                        "error_type": "JSON_PARSE"}

        if not response.ok:
            error_msg = f"HTTP {response.status_code} {response.reason}"
            if isinstance(data, dict) and "message" in data:
                error_msg = f"{error_msg}: {data['message']}"
            return {"success": False, "status": response.status_code,
                    "data": data, "error": error_msg, "error_type": "HTTP_ERROR"}

        return {"success": True, "status": response.status_code,
                "data": data, "error": None, "error_type": None}

    def get(self, path, **kwargs):
        """GET request."""
        return self._request("GET", path, **kwargs)

    def post(self, path, **kwargs):
        """POST request."""
        return self._request("POST", path, **kwargs)

    def put(self, path, **kwargs):
        """PUT request."""
        return self._request("PUT", path, **kwargs)

    def patch(self, path, **kwargs):
        """PATCH request."""
        return self._request("PATCH", path, **kwargs)

    def delete(self, path, **kwargs):
        """DELETE request."""
        return self._request("DELETE", path, **kwargs)

    def close(self):
        self.session.close()


api = APIClient("https://jsonplaceholder.typicode.com")

print("=== CRUD con JSONPlaceholder ===\n")

print("CREATE:")
r = api.post("/posts", json={"title": "Nuevo post", "body": "Contenido", "userId": 1})
print(f"  {'✅' if r['success'] else '❌'} {r['status']} → ID: {r['data']['id']}")

print("\nREAD:")
r = api.get("/posts/1")
print(f"  {'✅' if r['success'] else '❌'} {r['status']} → {r['data']['title'][:40]}")

print("\nUPDATE:")
r = api.patch("/posts/1", json={"title": "Título actualizado"})
print(f"  {'✅' if r['success'] else '❌'} {r['status']} → {r['data']['title']}")

print("\nDELETE:")
r = api.delete("/posts/1")
print(f"  {'✅' if r['success'] else '❌'} {r['status']}")

print("\nERROR:")
r = api.get("/posts/99999")
print(f"  {'✅' if r['success'] else '❌'} {r['status']} → {r.get('error', 'sin data')}")

api.close()

Output esperado:

=== CRUD con JSONPlaceholder ===

CREATE:
  ✅ 201 → ID: 101

READ:
  ✅ 200 → sunt aut facere repellat provident MDunt

UPDATE:
  ✅ 200 → Título actualizado

DELETE:
  ✅ 200

ERROR:
  ✅ 200 → sin data

Comparación: funciones sueltas vs APIClient

Aspecto              │ Funciones sueltas              │ APIClient (clase)
─────────────────────┼────────────────────────────────┼────────────────────────────
Configuración        │ Repetida en cada llamada       │ Una vez en __init__
Headers              │ Pasados manualmente            │ Persistentes en Session
Conexiones           │ Nueva TCP cada vez             │ Reutilizadas (keep-alive)
Error handling       │ Duplicado por función          │ Centralizado en _request()
Base URL             │ Concatenación manual           │ Automática con _build_url()
Autenticación        │ Header en cada petición        │ Configurada una vez
Testing              │ Difícil de mockear             │ Fácil: mockeas la clase
Mantenimiento        │ Cambio → N archivos            │ Cambio → 1 archivo

Cuándo usar cada patrón

Funciones sueltas: 1-3 peticiones a la misma API, scripts rápidos, prototipos
APIClient (clase):  4+ peticiones, múltiples APIs, código que se mantiene

Usando el APIClient con múltiples APIs

El poder del patrón se ve cuando consumes más de una API:

import requests
import json

class APIClient:
    """Cliente HTTP reutilizable completo."""

    def __init__(self, base_url, timeout=10, token=None):
        self.base_url = base_url.rstrip("/")
        self.timeout = timeout
        self.session = requests.Session()
        self.session.headers.update({
            "Accept": "application/json",
            "User-Agent": "PythonAPIClient/1.0"
        })
        if token:
            self.session.headers["Authorization"] = f"Bearer {token}"

    def _build_url(self, path):
        if path.startswith("http"):
            return path
        return f"{self.base_url}/{path.lstrip('/')}"

    def _request(self, method, path, **kwargs):
        url = self._build_url(path)
        kwargs.setdefault("timeout", self.timeout)
        try:
            response = self.session.request(method, url, **kwargs)
        except requests.exceptions.ConnectionError:
            return {"success": False, "status": None, "data": None,
                    "error": f"No se pudo conectar", "error_type": "CONNECTION"}
        except requests.exceptions.Timeout:
            return {"success": False, "status": None, "data": None,
                    "error": "Timeout", "error_type": "TIMEOUT"}
        except requests.exceptions.RequestException as e:
            return {"success": False, "status": None, "data": None,
                    "error": str(e), "error_type": "NETWORK"}

        if response.status_code == 204:
            return {"success": True, "status": 204, "data": None,
                    "error": None, "error_type": None}

        content_type = response.headers.get("Content-Type", "")
        data = None
        if "application/json" in content_type and response.text:
            try:
                data = response.json()
            except (json.JSONDecodeError, requests.exceptions.JSONDecodeError):
                return {"success": False, "status": response.status_code,
                        "data": None, "error": "JSON malformado",
                        "error_type": "JSON_PARSE"}

        if not response.ok:
            error_msg = f"HTTP {response.status_code}"
            if isinstance(data, dict) and "message" in data:
                error_msg += f": {data['message']}"
            return {"success": False, "status": response.status_code,
                    "data": data, "error": error_msg, "error_type": "HTTP_ERROR"}

        return {"success": True, "status": response.status_code,
                "data": data, "error": None, "error_type": None}

    def get(self, path, **kwargs):
        return self._request("GET", path, **kwargs)

    def post(self, path, **kwargs):
        return self._request("POST", path, **kwargs)

    def put(self, path, **kwargs):
        return self._request("PUT", path, **kwargs)

    def patch(self, path, **kwargs):
        return self._request("PATCH", path, **kwargs)

    def delete(self, path, **kwargs):
        return self._request("DELETE", path, **kwargs)

    def close(self):
        self.session.close()


github = APIClient("https://api.github.com")
jsonph = APIClient("https://jsonplaceholder.typicode.com")

print("=== GitHub API ===")
r = github.get("/users/octocat")
if r["success"]:
    user = r["data"]
    print(f"  User: {user['login']}")
    print(f"  Name: {user.get('name', 'N/A')}")
    print(f"  Repos: {user['public_repos']}")

r = github.get("/users/octocat/repos", params={"per_page": 3, "sort": "updated"})
if r["success"]:
    print(f"\n  Últimos 3 repos:")
    for repo in r["data"]:
        print(f"    - {repo['name']}: {repo.get('description') or 'Sin descripción'}")

print("\n=== JSONPlaceholder API ===")
r = jsonph.get("/users/1")
if r["success"]:
    u = r["data"]
    print(f"  User: {u['name']}")
    print(f"  Email: {u['email']}")
    print(f"  City: {u['address']['city']}")

r = jsonph.get("/users/1/posts", params={"_limit": 3})
if r["success"]:
    print(f"\n  Últimos 3 posts:")
    for post in r["data"]:
        print(f"    - {post['title'][:45]}")

github.close()
jsonph.close()

Output esperado:

=== GitHub API ===
  User: octocat
  Name: The Octocat
  Repos: 8

  Últimos 3 repos:
    - boysenberry-repo-1: Testing
    - git-consortium: This repo is for demonstration purposes.
    - hello-worId: My first repository on GitHub!

=== JSONPlaceholder API ===
  User: Leanne Graham
  Email: Sincere@april.biz
  City: Gwenborough

  Últimos 3 posts:
    - sunt aut facere repellat provident occaeca
    - qui est esse
    - ea molestias quasi exercitationem repellat

Cada API tiene su propia instancia de APIClient. Las sesiones son independientes. Los headers de GitHub (con vnd.github.v3+json) no interfieren con JSONPlaceholder.

Separación de concerns: HTTP vs lógica de negocio

El APIClient maneja HTTP. Tu código de aplicación maneja la lógica de negocio. Nunca mezcles ambos:

import requests
import json


class APIClient:
    def __init__(self, base_url, timeout=10, token=None):
        self.base_url = base_url.rstrip("/")
        self.timeout = timeout
        self.session = requests.Session()
        self.session.headers.update({"Accept": "application/json",
                                     "User-Agent": "PythonAPIClient/1.0"})
        if token:
            self.session.headers["Authorization"] = f"Bearer {token}"

    def _build_url(self, path):
        if path.startswith("http"):
            return path
        return f"{self.base_url}/{path.lstrip('/')}"

    def _request(self, method, path, **kwargs):
        url = self._build_url(path)
        kwargs.setdefault("timeout", self.timeout)
        try:
            response = self.session.request(method, url, **kwargs)
        except requests.exceptions.ConnectionError:
            return {"success": False, "status": None, "data": None,
                    "error": "No se pudo conectar", "error_type": "CONNECTION"}
        except requests.exceptions.Timeout:
            return {"success": False, "status": None, "data": None,
                    "error": "Timeout", "error_type": "TIMEOUT"}
        except requests.exceptions.RequestException as e:
            return {"success": False, "status": None, "data": None,
                    "error": str(e), "error_type": "NETWORK"}
        data = None
        if response.status_code != 204:
            ct = response.headers.get("Content-Type", "")
            if "application/json" in ct and response.text:
                try:
                    data = response.json()
                except Exception:
                    return {"success": False, "status": response.status_code,
                            "data": None, "error": "JSON malformado",
                            "error_type": "JSON_PARSE"}
        if not response.ok:
            msg = f"HTTP {response.status_code}"
            if isinstance(data, dict) and "message" in data:
                msg += f": {data['message']}"
            return {"success": False, "status": response.status_code,
                    "data": data, "error": msg, "error_type": "HTTP_ERROR"}
        return {"success": True, "status": response.status_code,
                "data": data, "error": None, "error_type": None}

    def get(self, path, **kwargs):
        return self._request("GET", path, **kwargs)

    def post(self, path, **kwargs):
        return self._request("POST", path, **kwargs)

    def close(self):
        self.session.close()


def get_github_profile(client, username):
    """Lógica de negocio: obtiene y formatea un perfil de GitHub."""
    result = client.get(f"/users/{username}")

    if not result["success"]:
        return f"Error al obtener @{username}: {result['error']}"

    user = result["data"]
    return {
        "username": user["login"],
        "display_name": user.get("name") or user["login"],
        "bio": user.get("bio") or "Sin biografía",
        "repos": user.get("public_repos", 0),
        "followers": user.get("followers", 0),
        "profile_url": user.get("html_url", ""),
    }


def get_user_top_repos(client, username, limit=3):
    """Lógica de negocio: obtiene los repos más populares."""
    result = client.get(f"/users/{username}/repos",
                        params={"sort": "stars", "per_page": limit})

    if not result["success"]:
        return []

    return [
        {
            "name": repo["name"],
            "stars": repo.get("stargazers_count", 0),
            "language": repo.get("language") or "N/A",
            "description": repo.get("description") or "Sin descripción",
        }
        for repo in result["data"]
    ]


github = APIClient("https://api.github.com")

profile = get_github_profile(github, "octocat")
if isinstance(profile, dict):
    print(f"👤 {profile['display_name']} (@{profile['username']})")
    print(f"   {profile['bio']}")
    print(f"   {profile['repos']} repos, {profile['followers']} followers")

    repos = get_user_top_repos(github, "octocat")
    if repos:
        print(f"\n   Top repos:")
        for repo in repos:
            print(f"   ⭐ {repo['stars']:>3} | {repo['name']:20s} | {repo['language']}")
else:
    print(profile)

github.close()

Output esperado:

👤 The Octocat (@octocat)
   Sin biografía
   8 repos, 0 followers

   Top repos:
   ⭐   0 | boysenberry-repo-1   | N/A
   ⭐   0 | git-consortium       | N/A
   ⭐   0 | hello-worId          | N/A

Fíjate en la separación:

  • ✅ APIClient solo sabe de HTTP: métodos, URLs, headers, errores de red
  • ✅ get_github_profile() solo sabe de GitHub: qué campos extraer, cómo formatear, qué defaults usar
  • ✅ El código principal solo sabe de la interfaz de usuario: qué mostrar y en qué formato

Si mañana GitHub cambia la estructura de su JSON, solo modificas get_github_profile(). El APIClient no cambia. Y si mañana quieres cambiar requests por httpx, solo modificas APIClient. Las funciones de negocio no cambian.

Context manager: with APIClient

Para garantizar que la sesión se cierra incluso si tu código lanza una excepción, implementa el protocolo context manager:

import requests
import json

class APIClient:
    def __init__(self, base_url, timeout=10, token=None):
        self.base_url = base_url.rstrip("/")
        self.timeout = timeout
        self.session = requests.Session()
        self.session.headers.update({"Accept": "application/json",
                                     "User-Agent": "PythonAPIClient/1.0"})
        if token:
            self.session.headers["Authorization"] = f"Bearer {token}"

    def _build_url(self, path):
        if path.startswith("http"):
            return path
        return f"{self.base_url}/{path.lstrip('/')}"

    def _request(self, method, path, **kwargs):
        url = self._build_url(path)
        kwargs.setdefault("timeout", self.timeout)
        try:
            response = self.session.request(method, url, **kwargs)
        except requests.exceptions.RequestException as e:
            return {"success": False, "status": None, "data": None,
                    "error": str(e), "error_type": "NETWORK"}
        data = None
        if response.status_code != 204:
            ct = response.headers.get("Content-Type", "")
            if "application/json" in ct and response.text:
                try:
                    data = response.json()
                except Exception:
                    data = None
        if not response.ok:
            msg = f"HTTP {response.status_code}"
            return {"success": False, "status": response.status_code,
                    "data": data, "error": msg, "error_type": "HTTP_ERROR"}
        return {"success": True, "status": response.status_code,
                "data": data, "error": None, "error_type": None}

    def get(self, path, **kwargs):
        return self._request("GET", path, **kwargs)

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.session.close()
        return False

    def close(self):
        self.session.close()


with APIClient("https://api.github.com") as github:
    result = github.get("/users/octocat")
    if result["success"]:
        print(f"User: {result['data']['login']}")

print("Session cerrada automáticamente")

Output esperado:

User: octocat
Session cerrada automáticamente

__enter__ devuelve self para usar as github. __exit__ cierra la sesión pase lo que pase. return False significa que no suprime excepciones — si tu código falla, el error se propaga después de cerrar la sesión.

Conexión con el proyecto

El APIClient que construiste en esta cápsula es exactamente la clase que vas a usar en la cápsula 08 (proyecto del módulo) y en el REST Client CLI del Módulo 4.

En la cápsula 08, vas a usar APIClient para construir un cliente que consume GitHub y JSONPlaceholder con error handling robusto. La estructura será:

api_client.py          → La clase APIClient (esta cápsula)
proyecto_modulo3.py    → Script que usa APIClient para consumir 2 APIs

En el Módulo 4, la estructura crece:

api_client.py          → APIClient (reutilizado del Módulo 3)
github_service.py      → Funciones de negocio para GitHub
jsonph_service.py      → Funciones de negocio para JSONPlaceholder
weather_service.py     → Funciones de negocio para OpenWeather
cli.py                 → Interfaz de línea de comandos

Cada servicio crea su instancia de APIClient con su URL base y configuración. El CLI solo llama a las funciones de negocio. La separación de concerns permite agregar una nueva API cambiando solo 2 cosas: crear un nuevo servicio y agregar un comando al CLI.

Troubleshooting

Problema 1: La sesión no envía los headers que configuré

Causa: Actualizaste los headers después de hacer la primera petición, o usaste self.session.headers = {...} en vez de .update(). El = reemplaza el dict completo, incluyendo headers que requests agrega automáticamente.

Solución:

# ❌ Reemplaza todo — pierde Accept-Encoding, Connection, etc.
self.session.headers = {"Authorization": "Bearer token"}

# ✅ Agrega sin borrar
self.session.headers.update({"Authorization": "Bearer token"})

# ✅ O asigna un header específico
self.session.headers["Authorization"] = "Bearer token"

Problema 2: URL con doble slash (https://api.com//users)

Causa: El base_url termina con / y el path empieza con /.

Solución: _build_url() usa rstrip("/") en la base y lstrip("/") en el path:

def _build_url(self, path):
    return f"{self.base_url}/{path.lstrip('/')}"

Problema 3: La sesión no se cierra y el programa cuelga al final

Causa: No llamaste client.close() y hay conexiones TCP abiertas que Python intenta cerrar al salir.

Solución: Usa with (context manager) o llama close() en un finally:

client = APIClient("https://api.github.com")
try:
    result = client.get("/users/octocat")
finally:
    client.close()

Problema 4: _request() no pasa mis parámetros a requests

Causa: Estás pasando parámetros posicionales en vez de keyword arguments. Los métodos del APIClient usan **kwargs.

Solución:

# ❌ Posicional — kwargs no incluye json
client.post("/posts", {"title": "test"})

# ✅ Keyword — kwargs incluye json=
client.post("/posts", json={"title": "test"})

Ejercicios

Ejercicio 1: APIClient básico (Fácil)

Crea un APIClient para https://jsonplaceholder.typicode.com y haz un CRUD completo: crear un post, leerlo, actualizarlo con PATCH, y eliminarlo. Muestra el resultado de cada operación.

Ver solución 
import requests
import json

class APIClient:
    def __init__(self, base_url, timeout=10):
        self.base_url = base_url.rstrip("/")
        self.timeout = timeout
        self.session = requests.Session()
        self.session.headers.update({"Accept": "application/json"})

    def _build_url(self, path):
        return f"{self.base_url}/{path.lstrip('/')}"

    def _request(self, method, path, **kwargs):
        kwargs.setdefault("timeout", self.timeout)
        try:
            r = self.session.request(method, self._build_url(path), **kwargs)
        except requests.exceptions.RequestException as e:
            return {"success": False, "data": None, "error": str(e)}
        data = None
        if r.text and "json" in r.headers.get("Content-Type", ""):
            try:
                data = r.json()
            except Exception:
                pass
        if not r.ok:
            return {"success": False, "data": data,
                    "error": f"HTTP {r.status_code}"}
        return {"success": True, "data": data, "error": None}

    def get(self, path, **kw):    return self._request("GET", path, **kw)
    def post(self, path, **kw):   return self._request("POST", path, **kw)
    def patch(self, path, **kw):  return self._request("PATCH", path, **kw)
    def delete(self, path, **kw): return self._request("DELETE", path, **kw)
    def close(self):              self.session.close()


api = APIClient("https://jsonplaceholder.typicode.com")

r = api.post("/posts", json={"title": "Mi post", "body": "Contenido", "userId": 1})
print(f"CREATE: {'✅' if r['success'] else '❌'} ID={r['data']['id']}")

r = api.get("/posts/1")
print(f"READ:   {'✅' if r['success'] else '❌'} {r['data']['title'][:35]}")

r = api.patch("/posts/1", json={"title": "Actualizado"})
print(f"UPDATE: {'✅' if r['success'] else '❌'} {r['data']['title']}")

r = api.delete("/posts/1")
print(f"DELETE: {'✅' if r['success'] else '❌'}")

api.close()

Explicación: Los 4 métodos (post, get, patch, delete) son wrappers de _request(). Cada uno pasa el método HTTP correcto y delega el error handling al método central.

Ejercicio 2: Multi-API client (Medio)

Crea dos instancias de APIClient — una para GitHub y otra para JSONPlaceholder. Obtén los 3 últimos repos de octocat y los 3 primeros posts de JSONPlaceholder. Muestra todo junto en un reporte formateado.

Ver solución 
import requests
import json

class APIClient:
    def __init__(self, base_url, timeout=10):
        self.base_url = base_url.rstrip("/")
        self.timeout = timeout
        self.session = requests.Session()
        self.session.headers.update({"Accept": "application/json"})

    def _request(self, method, path, **kwargs):
        url = f"{self.base_url}/{path.lstrip('/')}"
        kwargs.setdefault("timeout", self.timeout)
        try:
            r = self.session.request(method, url, **kwargs)
        except requests.exceptions.RequestException as e:
            return {"success": False, "data": None, "error": str(e)}
        data = None
        if r.text and "json" in r.headers.get("Content-Type", ""):
            try:
                data = r.json()
            except Exception:
                pass
        if not r.ok:
            return {"success": False, "data": data,
                    "error": f"HTTP {r.status_code}"}
        return {"success": True, "data": data, "error": None}

    def get(self, path, **kw): return self._request("GET", path, **kw)
    def close(self): self.session.close()


github = APIClient("https://api.github.com")
jsonph = APIClient("https://jsonplaceholder.typicode.com")

print("=" * 55)
print("  REPORTE MULTI-API")
print("=" * 55)

print("\n📦 GitHub — Repos de @octocat:")
r = github.get("/users/octocat/repos", params={"per_page": 3, "sort": "updated"})
if r["success"]:
    for repo in r["data"]:
        desc = repo.get("description") or "Sin descripción"
        lang = repo.get("language") or "N/A"
        print(f"  [{lang:10s}] {repo['name']:25s} — {desc[:35]}")
else:
    print(f"  Error: {r['error']}")

print("\n📝 JSONPlaceholder — Últimos posts:")
r = jsonph.get("/posts", params={"_limit": 3})
if r["success"]:
    for post in r["data"]:
        print(f"  #{post['id']:>3} | {post['title'][:45]}")
else:
    print(f"  Error: {r['error']}")

print("\n" + "=" * 55)

github.close()
jsonph.close()

Explicación: Cada API tiene su propia instancia con su base URL. Las sesiones son independientes — los headers de una no afectan a la otra. Este es exactamente el patrón que usarás en el Módulo 4 con 5+ APIs.

Ejercicio 3: APIClient con context manager (Medio)

Agrega __enter__ y __exit__ al APIClient para soportar with. Úsalo para hacer 3 peticiones a la API de GitHub dentro de un bloque with. Verifica que la sesión se cierra al salir del bloque.

Ver solución 
import requests
import json

class APIClient:
    def __init__(self, base_url, timeout=10):
        self.base_url = base_url.rstrip("/")
        self.timeout = timeout
        self.session = requests.Session()
        self.session.headers.update({"Accept": "application/json"})

    def _request(self, method, path, **kwargs):
        url = f"{self.base_url}/{path.lstrip('/')}"
        kwargs.setdefault("timeout", self.timeout)
        try:
            r = self.session.request(method, url, **kwargs)
        except requests.exceptions.RequestException as e:
            return {"success": False, "data": None, "error": str(e)}
        data = None
        if r.text and "json" in r.headers.get("Content-Type", ""):
            try:
                data = r.json()
            except Exception:
                pass
        if not r.ok:
            return {"success": False, "data": data,
                    "error": f"HTTP {r.status_code}"}
        return {"success": True, "data": data, "error": None}

    def get(self, path, **kw): return self._request("GET", path, **kw)

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.session.close()
        return False

    def close(self):
        self.session.close()


with APIClient("https://api.github.com") as github:
    users = ["octocat", "mojombo", "defunkt"]

    for username in users:
        r = github.get(f"/users/{username}")
        if r["success"]:
            u = r["data"]
            name = u.get("name") or u["login"]
            print(f"  @{u['login']:15s} → {name} ({u['public_repos']} repos)")
        else:
            print(f"  @{username:15s} → Error: {r['error']}")

print("\nSession cerrada automáticamente al salir del with")

Explicación: __enter__ devuelve self, __exit__ llama close(). El patrón with garantiza que la sesión se cierra aunque tu código lance una excepción.

Ejercicio 4: APIClient con logging de peticiones (Difícil)

Extiende APIClient para que lleve un registro (log) de todas las peticiones hechas: URL, método, status, tiempo en ms, y si fue exitosa. Agrega un método .stats() que muestre un resumen: total de peticiones, exitosas, fallidas, tiempo promedio.

Ver solución 
import requests
import json
import time

class APIClient:
    def __init__(self, base_url, timeout=10):
        self.base_url = base_url.rstrip("/")
        self.timeout = timeout
        self.session = requests.Session()
        self.session.headers.update({"Accept": "application/json"})
        self._log = []

    def _request(self, method, path, **kwargs):
        url = f"{self.base_url}/{path.lstrip('/')}"
        kwargs.setdefault("timeout", self.timeout)

        start = time.time()
        try:
            r = self.session.request(method, url, **kwargs)
            ms = (time.time() - start) * 1000
        except requests.exceptions.RequestException as e:
            ms = (time.time() - start) * 1000
            self._log.append({"method": method, "path": path,
                              "status": None, "ms": ms, "success": False})
            return {"success": False, "data": None, "error": str(e)}

        data = None
        if r.text and "json" in r.headers.get("Content-Type", ""):
            try:
                data = r.json()
            except Exception:
                pass

        success = r.ok
        self._log.append({"method": method, "path": path,
                          "status": r.status_code, "ms": ms, "success": success})

        if not success:
            return {"success": False, "data": data,
                    "error": f"HTTP {r.status_code}"}
        return {"success": True, "data": data, "error": None}

    def get(self, path, **kw): return self._request("GET", path, **kw)
    def post(self, path, **kw): return self._request("POST", path, **kw)

    def stats(self):
        total = len(self._log)
        ok = sum(1 for e in self._log if e["success"])
        fail = total - ok
        avg_ms = sum(e["ms"] for e in self._log) / total if total else 0

        print(f"\n{'=' * 55}")
        print(f"  API STATS — {self.base_url}")
        print(f"{'=' * 55}")
        print(f"  Total:    {total} peticiones")
        print(f"  Exitosas: {ok}")
        print(f"  Fallidas: {fail}")
        print(f"  Promedio: {avg_ms:.0f}ms")
        print(f"\n  Detalle:")
        for entry in self._log:
            status = entry['status'] or 'ERR'
            icon = '✅' if entry['success'] else '❌'
            print(f"  {icon} {entry['method']:6s} {entry['path']:30s} "
                  f"{status:>5} {entry['ms']:>6.0f}ms")

    def close(self):
        self.session.close()


api = APIClient("https://api.github.com")

api.get("/users/octocat")
api.get("/users/mojombo")
api.get("/users/no-existe-xyz-99999")
api.get("/users/defunkt")

api.stats()
api.close()

Explicación: self._log acumula un dict por cada petición con metadata. stats() calcula y muestra un resumen. Este patrón de logging es útil para profiling — te dice cuánto tardan las peticiones y dónde están los errores.

Ejercicio 5: Verificador de salud de APIs (Difícil)

Escribe una función health_check(apis) que reciba un dict de {nombre: url} y cree un APIClient temporal para cada una, haga un GET a la URL base, y reporte: nombre, status code, tiempo, y si está "saludable" (200 + JSON). Pruébala con al menos 4 URLs.

Ver solución 
import requests
import json
import time

class APIClient:
    def __init__(self, base_url, timeout=10):
        self.base_url = base_url.rstrip("/")
        self.timeout = timeout
        self.session = requests.Session()
        self.session.headers.update({"Accept": "application/json"})

    def get(self, path="/", **kwargs):
        url = f"{self.base_url}/{path.lstrip('/')}"
        kwargs.setdefault("timeout", self.timeout)
        start = time.time()
        try:
            r = self.session.request("GET", url, **kwargs)
            ms = (time.time() - start) * 1000
        except requests.exceptions.RequestException as e:
            ms = (time.time() - start) * 1000
            return {"status": None, "ms": ms, "healthy": False,
                    "reason": str(type(e).__name__)}

        is_json = "json" in r.headers.get("Content-Type", "")
        healthy = r.ok and is_json

        reason = "OK" if healthy else ""
        if not r.ok:
            reason = f"HTTP {r.status_code}"
        elif not is_json:
            reason = "No JSON"

        return {"status": r.status_code, "ms": ms,
                "healthy": healthy, "reason": reason}

    def close(self):
        self.session.close()


def health_check(apis):
    """Verifica la salud de múltiples APIs."""
    print(f"\n{'=' * 65}")
    print(f"  API HEALTH CHECK")
    print(f"{'=' * 65}")
    print(f"  {'Nombre':<20} {'Status':>7} {'Tiempo':>8} {'Estado':<15}")
    print(f"  {'-'*20} {'-'*7} {'-'*8} {'-'*15}")

    results = []
    for name, url in apis.items():
        client = APIClient(url, timeout=5)
        result = client.get("/")
        client.close()

        status_str = str(result['status']) if result['status'] else 'ERR'
        icon = '✅' if result['healthy'] else '❌'
        print(f"  {name:<20} {status_str:>7} {result['ms']:>6.0f}ms "
              f"{icon} {result['reason']}")
        results.append({**result, "name": name})

    healthy = sum(1 for r in results if r["healthy"])
    print(f"\n  Resultado: {healthy}/{len(results)} APIs saludables")
    print(f"{'=' * 65}")


health_check({
    "GitHub API":        "https://api.github.com",
    "JSONPlaceholder":   "https://jsonplaceholder.typicode.com",
    "httpbin":           "https://httpbin.org",
    "No existe":         "https://no-existe-xyz-99999.invalid",
})

Explicación: Cada API se verifica de forma independiente con su propio APIClient temporal. El health check prueba conexión, status code y Content-Type. Este patrón es útil para dashboards de monitoreo y para el startup del REST Client CLI.

Resumen

  • requests.Session centraliza headers y reutiliza conexiones TCP — úsalo siempre que hagas 2+ peticiones al mismo host
  • APIClient encapsula Session, base URL, timeout, auth y error handling en una clase reutilizable
  • _request() es el método privado que centraliza toda la lógica HTTP — los métodos públicos son wrappers limpios
  • _build_url() combina base URL + path evitando doble slash
  • Separación de concerns: APIClient maneja HTTP, las funciones de negocio manejan datos, el código principal maneja la UI
  • Context manager (with) garantiza que la sesión se cierra aunque haya excepciones
  • Una instancia por API: GitHub tiene su APIClient, JSONPlaceholder tiene el suyo, cada uno con su configuración
  • Este patrón es la base del REST Client CLI del Módulo 4 — lo escribes una vez, lo usas con 5+ APIs

Próxima cápsula: Proyecto del Módulo 3 — vas a usar APIClient para construir un cliente robusto que consume GitHub y JSONPlaceholder con error handling completo y output formateado.

Recursos adicionales

  1. Requests: Session Objects — Documentación oficial de Session, cookies persistentes y transport adapters
  2. Requests: Prepared Requests — Control avanzado sobre cómo se construyen los requests
  3. Real Python: Python's requests Library (Advanced) — Tutorial de Session con ejemplos prácticos
  4. Python Data Model: Context Managers — Documentación oficial de enter y exit
  5. GitHub REST API — API pública para testing del APIClient
  6. JSONPlaceholder — Guía de la API de pruebas para desarrollo frontend y backend
← AnteriorError Handling para JSON y APIsSiguiente →Proyecto del Módulo 3: API Client Robusto