"""Cliente HTTP configurable hacia la API externa de CRM / datos. La base se arma con ``CRM_API_BASE_ID`` (ej. project id de MockAPI) o con ``CRM_API_BASE_URL`` absoluta. Los paths relativos se configuran por recurso (``CRM_ENDPOINT_CLIENTES``, ``CRM_ENDPOINT_PRODUCTOS``, ``CRM_ENDPOINT_POLIZAS``, …). """ from __future__ import annotations import logging from typing import Any from urllib.parse import quote, urljoin import httpx from crm.models import Cliente, Producto, Polizas, ResourceResult logger = logging.getLogger(__name__) class CrmApiError(RuntimeError): def __init__( self, message: str, *, status_code: int | None = None, path: str = "", body: Any = None, ) -> None: super().__init__(message) self.status_code = status_code self.path = path self.body = body class CrmApiClient: """Consulta recursos REST de la API externa (clientes, productos, polizas, etc.).""" def __init__( self, *, base_url: str, endpoints: dict[str, str] | None = None, api_key: str = "", timeout: float = 15.0, enabled: bool = True, ) -> None: self.base_url = base_url.rstrip("/") + "/" self.endpoints = { "clientes": "clientes", "productos": "productos", "polizas": "polizas", **(endpoints or {}), } self.api_key = (api_key or "").strip() self.enabled = enabled headers: dict[str, str] = { "Accept": "application/json", "Content-Type": "application/json", } if self.api_key: headers["Authorization"] = f"Bearer {self.api_key}" self._client = httpx.AsyncClient( base_url=self.base_url, headers=headers, timeout=timeout, ) logger.info( "CRM API client base=%s endpoints=%s enabled=%s", self.base_url, self.endpoints, self.enabled, ) def resolve_path(self, resource: str, *parts: str) -> str: """Arma path relativo: resource key → endpoint configurado + ids.""" base = self.endpoints.get(resource) or resource segments = [base.strip("/")] for part in parts: value = str(part).strip().strip("/") if value: segments.append(quote(value, safe="")) return "/".join(segments) async def request( self, method: str, path: str, *, params: dict[str, Any] | None = None, json: dict[str, Any] | None = None, resource: str = "", ) -> ResourceResult: if not self.enabled: raise CrmApiError("CRM API deshabilitada (CRM_API_ENABLED=false)") clean = path.lstrip("/") logger.debug("CRM %s %s params=%s", method.upper(), clean, params) try: response = await self._client.request( method.upper(), clean, params=params, json=json, ) except httpx.HTTPError as exc: raise CrmApiError( f"Error de red CRM {method.upper()} {clean}: {exc}", path=clean, ) from exc body: Any try: body = response.json() except ValueError: body = response.text result = ResourceResult( data=body, status_code=response.status_code, resource=resource, path=clean, ) if response.status_code == 404: return result if response.status_code >= 400: raise CrmApiError( f"CRM {method.upper()} {clean} → HTTP {response.status_code}", status_code=response.status_code, path=clean, body=body, ) return result async def get( self, resource: str, *parts: str, params: dict[str, Any] | None = None, ) -> ResourceResult: path = self.resolve_path(resource, *parts) return await self.request("GET", path, params=params, resource=resource) async def list( self, resource: str, *, params: dict[str, Any] | None = None, ) -> ResourceResult: return await self.get(resource, params=params) # --- Clientes ----------------------------------------------------------- async def list_clientes( self, *, params: dict[str, Any] | None = None ) -> list[Cliente]: result = await self.list("clientes", params=params) if result.status_code == 404 or result.is_empty: return [] rows = result.data if isinstance(result.data, list) else [result.data] return [ Cliente.from_api(row) for row in rows if isinstance(row, dict) ] async def get_cliente(self, cliente_id: str) -> Cliente | None: result = await self.get("clientes", cliente_id) if result.status_code == 404 or not isinstance(result.data, dict): return None return Cliente.from_api(result.data) async def find_cliente_by_documento(self, documento: str) -> Cliente | None: """Busca por id/documento. Primero GET /cliente/{doc}; si falla, lista.""" doc = str(documento).strip() if not doc: return None found = await self.get_cliente(doc) if found: return found # Fallback: algunos backends filtran por query (?id= / ?documento=) for params in ({"id": doc}, {"documento": doc}, {"search": doc}): try: matches = await self.list_clientes(params=params) except CrmApiError: continue for cliente in matches: if cliente.id == doc: return cliente if len(matches) == 1: return matches[0] return None # --- Productos ---------------------------------------------------------- async def list_productos( self, *, params: dict[str, Any] | None = None ) -> list[Producto]: result = await self.list("productos", params=params) if result.status_code == 404 or result.is_empty: return [] rows = result.data if isinstance(result.data, list) else [result.data] return [ Producto.from_api(row) for row in rows if isinstance(row, dict) ] async def get_producto(self, producto_id: str) -> Producto | None: result = await self.get("productos", producto_id) if result.status_code == 404 or not isinstance(result.data, dict): return None return Producto.from_api(result.data) # --- Genérico / pólizas ------------------------------------------------- async def get_poliza(self, poliza_id: str) -> Polizas | None: result = await self.get("polizas", poliza_id) if result.status_code == 404 or not isinstance(result.data, dict): return None return Polizas.from_api(result.data) async def list_polizas( self, *, params: dict[str, Any] | None = None ) -> list[Polizas]: result = await self.list("polizas", params=params) if result.status_code == 404 or result.is_empty: return [] rows = result.data if isinstance(result.data, list) else [result.data] return [Polizas.from_api(row) for row in rows if isinstance(row, dict)] def absolute_url(self, resource: str, *parts: str) -> str: return urljoin(self.base_url, self.resolve_path(resource, *parts)) async def close(self) -> None: await self._client.aclose()