142 lines
4.2 KiB
Python
142 lines
4.2 KiB
Python
|
|
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
|
|
|
|
################################################### |