import requests import urllib3 from config_loader import CONFIG ################################################### def post(session, recurso:str, json_body: dict = None): url = CONFIG.service_layer.url if not url.endswith('/'): url += '/' headers = { "accept": "application/json", "content-type": "application/json" } urllib3.disable_warnings() end_point = f"{url}{recurso}" try: response = None if json_body: response = session.post( end_point, headers = headers, json = json_body, verify = False ) else: response = session.post( end_point, headers = headers, verify = False ) response.raise_for_status() print(f" ✅ [POST] {recurso} - Status {response.status_code}") return response except requests.RequestException as ex: print(f" ❌ [POST] {recurso} - Falló: {ex}") return None ################################################### def get(session, recurso:str): SL_MAXPAGESIZE = CONFIG.service_layer.maxpagesize url = CONFIG.service_layer.url if not url.endswith('/'): url += '/' headers = { "accept": "application/json", "content-type": "application/json", "Prefer": f"odata.maxpagesize={SL_MAXPAGESIZE}" } urllib3.disable_warnings() end_point = f"{url}{recurso}" try: response = session.get( end_point, headers = headers, verify = False ) response.raise_for_status() print(f" ✅ [GET] {recurso} - Status {response.status_code}") return response except requests.RequestException as ex: print(f" ❌ [GET] {recurso} - Falló: {ex}") return None ################################################### def patch(session, recurso:str, json_body: dict): url = CONFIG.service_layer.url REPLACE_COLLECTIONS = CONFIG.service_layer.b1s_replacecollectionsonpatch if not url.endswith('/'): url += '/' headers = { "accept": "application/json", "content-type": "application/json" } if REPLACE_COLLECTIONS: headers["B1S-ReplaceCollectionsOnPatch"] = "true" urllib3.disable_warnings() end_point = f"{url}{recurso}" try: response = session.patch( end_point, headers = headers, json = json_body, verify = False ) if response.status_code == 204: print(f" ✅ [PATCH] {recurso} - Status {response.status_code}") return response if 400 <= response.status_code <= 499: print(f" ❌ [PATCH] {recurso} - Error HTTP {response.status_code}") try: error_body = response.json() import json as _json error = error_body.get("error", {}) mensaje = error.get("message", {}) if isinstance(mensaje, dict): mensaje = mensaje.get("value", str(mensaje)) codigo = error.get("code", "N/A") print(f" 📋 SAP Error Code : {codigo}") print(f" 📋 SAP Error Message: {mensaje}") # TransactionNotification (validaciones de UDF, triggers, etc.) notif = error_body.get("TransactionNotification") or error.get("innererror", {}).get("TransactionNotification") if notif: print(f" 📋 TransactionNotification: {notif}") except Exception: print(f" 📋 Respuesta SAP (raw): {response.text}") return None # Otros errores no esperados response.raise_for_status() return response except requests.RequestException as ex: print(f" ❌ [PATCH] {recurso} - Excepción: {ex}") return None ###################################################