99 lines
No EOL
3.5 KiB
Python
99 lines
No EOL
3.5 KiB
Python
import requests
|
|
from typing import Dict, List, Optional
|
|
from config import settings
|
|
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"
|
|
}
|
|
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
|
|
)
|
|
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}")
|
|
raise
|
|
time.sleep(2 ** attempt) # Backoff exponentiel
|
|
|
|
# === CLIENTS ===
|
|
def lister_clients(self, filtre: str = "") -> List[Dict]:
|
|
return self._post("/sage/clients/list", {"filtre": filtre}).get("data", [])
|
|
|
|
def lire_client(self, code: str) -> Optional[Dict]:
|
|
return self._post("/sage/clients/get", {"code": code}).get("data")
|
|
|
|
# === ARTICLES ===
|
|
def lister_articles(self, filtre: str = "") -> List[Dict]:
|
|
return self._post("/sage/articles/list", {"filtre": filtre}).get("data", [])
|
|
|
|
def lire_article(self, ref: str) -> Optional[Dict]:
|
|
return self._post("/sage/articles/get", {"code": ref}).get("data")
|
|
|
|
# === DEVIS ===
|
|
def creer_devis(self, devis_data: Dict) -> Dict:
|
|
return self._post("/sage/devis/create", devis_data).get("data", {})
|
|
|
|
def lire_devis(self, numero: str) -> Optional[Dict]:
|
|
return self._post("/sage/devis/get", {"code": numero}).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")
|
|
|
|
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
|
|
})
|
|
return resp.get("success", False)
|
|
|
|
# === CONTACTS ===
|
|
def lire_contact_client(self, code_client: str) -> Optional[Dict]:
|
|
return self._post("/sage/contact/read", {"code": code_client}).get("data")
|
|
|
|
# === CACHE ===
|
|
def refresh_cache(self) -> Dict:
|
|
return self._post("/sage/cache/refresh")
|
|
|
|
# === HEALTH ===
|
|
def health(self) -> dict:
|
|
try:
|
|
r = requests.get(f"{self.url}/health", timeout=5)
|
|
return r.json()
|
|
except:
|
|
return {"status": "down"}
|
|
|
|
# Instance globale
|
|
sage_client = SageGatewayClient() |