diff --git a/sage_client.py b/sage_client.py index a0b92fe..0b5cf95 100644 --- a/sage_client.py +++ b/sage_client.py @@ -5,38 +5,63 @@ import logging logger = logging.getLogger(__name__) + class SageGatewayClient: """ Client HTTP pour communiquer avec la gateway Sage Windows """ - + def __init__(self): self.url = settings.sage_gateway_url.rstrip("/") self.headers = { "X-Sage-Token": settings.sage_gateway_token, - "Content-Type": "application/json" + "Content-Type": "application/json", } self.timeout = 30 def _post(self, endpoint: str, data: dict = None, retries: int = 3) -> dict: """POST avec retry automatique""" import time - + for attempt in range(retries): try: r = requests.post( f"{self.url}{endpoint}", json=data or {}, headers=self.headers, - timeout=self.timeout + timeout=self.timeout, ) r.raise_for_status() return r.json() except requests.exceptions.RequestException as e: if attempt == retries - 1: - logger.error(f"❌ Échec après {retries} tentatives sur {endpoint}: {e}") + logger.error( + f"❌ Échec après {retries} tentatives sur {endpoint}: {e}" + ) raise - time.sleep(2 ** attempt) # Backoff exponentiel + time.sleep(2**attempt) + + def _get(self, endpoint: str, params: dict = None, retries: int = 3) -> dict: + """GET avec retry automatique""" + import time + + for attempt in range(retries): + try: + r = requests.get( + f"{self.url}{endpoint}", + params=params or {}, + headers=self.headers, + timeout=self.timeout, + ) + r.raise_for_status() + return r.json() + except requests.exceptions.RequestException as e: + if attempt == retries - 1: + logger.error( + f"❌ Échec GET après {retries} tentatives sur {endpoint}: {e}" + ) + raise + time.sleep(2**attempt) # === CLIENTS === def lister_clients(self, filtre: str = "") -> List[Dict]: @@ -59,34 +84,93 @@ class SageGatewayClient: def lire_devis(self, numero: str) -> Optional[Dict]: return self._post("/sage/devis/get", {"code": numero}).get("data") + # 🆕 US-A1: Lister devis + def lister_devis( + self, limit: int = 100, statut: Optional[int] = None + ) -> List[Dict]: + """Liste tous les devis avec filtres""" + payload = {"limit": limit} + if statut is not None: + payload["statut"] = statut + return self._post("/sage/devis/list", payload).get("data", []) + + # 🆕 US-A1: Changer statut devis + def changer_statut_devis(self, numero: str, nouveau_statut: int) -> Dict: + """Change le statut d'un devis""" + return self._post( + "/sage/devis/statut", {"numero": numero, "nouveau_statut": nouveau_statut} + ).get("data", {}) + # === DOCUMENTS === def lire_document(self, numero: str, type_doc: int) -> Optional[Dict]: - return self._post("/sage/documents/get", {"numero": numero, "type_doc": type_doc}).get("data") + return self._post( + "/sage/documents/get", {"numero": numero, "type_doc": type_doc} + ).get("data") - def transformer_document(self, numero_source: str, type_source: int, type_cible: int) -> Dict: - return self._post("/sage/documents/transform", { - "numero_source": numero_source, - "type_source": type_source, - "type_cible": type_cible - }).get("data", {}) + def transformer_document( + self, numero_source: str, type_source: int, type_cible: int + ) -> Dict: + return self._post( + "/sage/documents/transform", + { + "numero_source": numero_source, + "type_source": type_source, + "type_cible": type_cible, + }, + ).get("data", {}) - def mettre_a_jour_champ_libre(self, doc_id: str, type_doc: int, nom_champ: str, valeur: str) -> bool: - resp = self._post("/sage/documents/champ-libre", { - "doc_id": doc_id, - "type_doc": type_doc, - "nom_champ": nom_champ, - "valeur": valeur - }) + def mettre_a_jour_champ_libre( + self, doc_id: str, type_doc: int, nom_champ: str, valeur: str + ) -> bool: + resp = self._post( + "/sage/documents/champ-libre", + { + "doc_id": doc_id, + "type_doc": type_doc, + "nom_champ": nom_champ, + "valeur": valeur, + }, + ) return resp.get("success", False) + # 🆕 US-A2: Lister commandes + def lister_commandes( + self, limit: int = 100, statut: Optional[int] = None + ) -> List[Dict]: + """Liste toutes les commandes""" + payload = {"limit": limit} + if statut is not None: + payload["statut"] = statut + return self._post("/sage/commandes/list", payload).get("data", []) + + # 🆕 US-A7: Lister factures + def lister_factures( + self, limit: int = 100, statut: Optional[int] = None + ) -> List[Dict]: + """Liste toutes les factures""" + payload = {"limit": limit} + if statut is not None: + payload["statut"] = statut + return self._post("/sage/factures/list", payload).get("data", []) + # === CONTACTS === def lire_contact_client(self, code_client: str) -> Optional[Dict]: return self._post("/sage/contact/read", {"code": code_client}).get("data") + # 🆕 US-A5: Lire remise max client + def lire_remise_max_client(self, code_client: str) -> float: + """Récupère la remise max autorisée pour un client""" + result = self._post("/sage/client/remise-max", {"code": code_client}) + return result.get("data", {}).get("remise_max", 10.0) + # === CACHE === def refresh_cache(self) -> Dict: return self._post("/sage/cache/refresh") + def get_cache_info(self) -> Dict: + """Récupère les infos du cache""" + return self._get("/sage/cache/info").get("data", {}) + # === HEALTH === def health(self) -> dict: try: @@ -95,5 +179,6 @@ class SageGatewayClient: except: return {"status": "down"} + # Instance globale -sage_client = SageGatewayClient() \ No newline at end of file +sage_client = SageGatewayClient()