src/inventario/client.py"""Cliente HTTP del inventario.
Todo lo que puede fallar en la red pasa por aquí: timeouts, conexión rechazada,
respuestas de error. El resto del programa solo ve dos excepciones nuestras.
"""
import os
import time
import httpx
class InventarioAPIError(Exception):
"""La API respondió, pero con un error (4xx / 5xx)."""
def __init__(self, status: int, detail: str):
super().__init__(f"{status}: {detail}")
self.status = status
self.detail = detail
class InventarioNoDisponible(Exception):
"""No hubo respuesta: conexión rechazada, timeout, DNS. La API puede estar caída."""
# Errores de transporte: la petición no llegó o la respuesta no volvió.
_SIN_RESPUESTA = (httpx.TimeoutException, httpx.ConnectError)
class InventarioClient:
def __init__(
self,
base_url: str | None = None,
api_key: str | None = None,
timeout: float = 5.0,
reintentos: int = 3,
espera_base: float = 0.2,
):
self.base_url = (base_url or os.environ.get("INVENTARIO_API_URL", "")).rstrip("/")
if not self.base_url:
raise ValueError("falta base_url o la variable de entorno INVENTARIO_API_URL")
headers = {"X-API-Key": api_key} if api_key else {}
self._http = httpx.Client(base_url=self.base_url, timeout=timeout, headers=headers)
self.reintentos = reintentos
self.espera_base = espera_base
# --- transporte -------------------------------------------------------
def _get(self, path: str, params: dict | None = None) -> dict | list:
"""GET con reintentos: leer dos veces no cambia nada (idempotente)."""
ultimo: Exception | None = None
for intento in range(self.reintentos):
try:
respuesta = self._http.get(path, params=params)
except _SIN_RESPUESTA as exc:
ultimo = exc
time.sleep(self.espera_base * (2**intento)) # 0.2 s, 0.4 s, 0.8 s…
continue
return self._chequear(respuesta)
raise InventarioNoDisponible(
f"sin respuesta de {self.base_url} tras {self.reintentos} intentos"
) from ultimo
def _post(self, path: str, json: dict) -> dict:
"""POST sin reintentos: si el primero llegó y la respuesta se perdió, repetirlo duplica."""
try:
respuesta = self._http.post(path, json=json)
except _SIN_RESPUESTA as exc:
raise InventarioNoDisponible(f"sin respuesta de {self.base_url}") from exc
return self._chequear(respuesta)
@staticmethod
def _chequear(respuesta: httpx.Response) -> dict | list:
if respuesta.is_success:
return respuesta.json()
try:
detail = respuesta.json().get("detail", respuesta.text)
except ValueError:
detail = respuesta.text
raise InventarioAPIError(respuesta.status_code, str(detail))
# --- operaciones (espejo de la API del P4) -----------------------------
def salud(self) -> dict:
return self._get("/health")
def reactivos(self, q: str | None = None) -> list:
return self._get("/reactivos", params={"q": q} if q else None)
def crear_reactivo(self, nombre: str, unidad: str, cas: str | None = None) -> dict:
return self._post("/reactivos", {"nombre": nombre, "unidad": unidad, "cas": cas})
def lotes_por_vencer(self, dias: int = 30) -> list:
return self._get("/lotes/por-vencer", params={"dias": dias})
def registrar_salida(self, codigo_lote: str, cantidad: float, nota: str | None = None) -> dict:
return self._post(
"/movimientos", {"codigo_lote": codigo_lote, "cantidad": cantidad, "nota": nota}
)
# --- ciclo de vida -----------------------------------------------------
def close(self) -> None:
self._http.close()
def __enter__(self):
return self
def __exit__(self, *exc):
self.close()