#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Ejemplo de consumo de la API OData de Acqua ERP. Solo necesita `requests`. Hace el recorrido completo que conviene hacer al integrar: 1. Lee el documento de servicio para saber que entidades hay habilitadas. 2. Lee el $metadata y muestra campos, clave y relaciones de una entidad. 3. Consulta datos con $select, $filter y $expand. 4. Recorre todas las paginas siguiendo @odata.nextLink. Uso: pip install requests export SYNERGY_ODATA_ROOT="https://TU-HOST/synergy-server/jwt-odata-key/" export SYNERGY_API_KEY="TU_API_KEY" python3 ejemplo_odata.py # usa la primera entidad disponible python3 ejemplo_odata.py o_customers # o la que le indiques Documentacion completa: api-docs/odata-api.html del servidor. """ import os import sys import xml.etree.ElementTree as ET import requests ROOT = os.environ.get("SYNERGY_ODATA_ROOT", "https://TU-HOST/synergy-server/jwt-odata-key/") API_KEY = os.environ.get("SYNERGY_API_KEY", "") TIMEOUT = 60 # Namespaces del CSDL de OData v4 NS = { "edmx": "http://docs.oasis-open.org/odata/ns/edmx", "edm": "http://docs.oasis-open.org/odata/ns/edm", } def sesion(): """Sesion con la credencial y el Accept correctos. El Accept importa: la negociacion del servicio es estricta y un valor como text/plain devuelve 406 incluso donde la respuesta es texto. """ s = requests.Session() s.headers.update({ "X-API-Key": API_KEY, "Accept": "application/json", }) return s def pedir(s, url, params=None): """GET con manejo del formato de error de OData.""" r = s.get(url, params=params, timeout=TIMEOUT) if not r.ok: detalle = "" try: error = r.json().get("error", {}) detalle = "%s: %s" % (error.get("code", "?"), error.get("message", "")) for d in error.get("details", []): detalle += "\n - " + d.get("message", "") except ValueError: detalle = r.text[:400] raise SystemExit("HTTP %d en %s\n %s" % (r.status_code, r.url, detalle)) return r # --------------------------------------------------------------------- # 1. documento de servicio # --------------------------------------------------------------------- def entidades_disponibles(s): """Entity sets habilitados para este cliente. Es la fuente de verdad: lo que no aparece aca no esta habilitado, por mas que exista en el ERP. """ cuerpo = pedir(s, ROOT).json() return [ {"nombre": e["name"], "titulo": e.get("title", e["name"])} for e in cuerpo.get("value", []) ] # --------------------------------------------------------------------- # 2. metadata # --------------------------------------------------------------------- def describir(s, entity_set): """Campos, clave y relaciones de una entidad, leidos del $metadata.""" xml = pedir(s, ROOT + "$metadata").text raiz = ET.fromstring(xml) tipo = None for candidato in raiz.iter("{%s}EntityType" % NS["edm"]): if candidato.get("Name") == entity_set: tipo = candidato break if tipo is None: return None def descripcion(nodo): for anotacion in nodo.findall("{%s}Annotation" % NS["edm"]): if anotacion.get("Term") in ("Core.Description", "Org.OData.Core.V1.Description"): return anotacion.get("String") return None claves = [ ref.get("Name") for clave in tipo.findall("{%s}Key" % NS["edm"]) for ref in clave.findall("{%s}PropertyRef" % NS["edm"]) ] propiedades = [] for prop in tipo.findall("{%s}Property" % NS["edm"]): propiedades.append({ "nombre": prop.get("Name"), "tipo": prop.get("Type"), "nullable": prop.get("Nullable") != "false", "largo": prop.get("MaxLength"), # la etiqueta del ERP, en espanol: es lo que conviene mostrar "etiqueta": descripcion(prop), "clave": prop.get("Name") in claves, }) navegaciones = [] for nav in tipo.findall("{%s}NavigationProperty" % NS["edm"]): tipo_nav = nav.get("Type", "") coleccion = tipo_nav.startswith("Collection(") destino = tipo_nav[len("Collection("):-1] if coleccion else tipo_nav navegaciones.append({ "nombre": nav.get("Name"), "destino": destino.split(".")[-1], "coleccion": coleccion, "etiqueta": descripcion(nav), }) return { "etiqueta": descripcion(tipo), "claves": claves, "propiedades": propiedades, "navegaciones": navegaciones, } # --------------------------------------------------------------------- # 3 y 4. datos, con paginacion # --------------------------------------------------------------------- def recorrer(s, entity_set, **opciones): """Itera TODAS las filas, siguiendo @odata.nextLink. Las opciones se pasan como palabras clave sin el signo: select="a,b" se manda como $select=a,b. """ url = ROOT + entity_set params = {"$" + k: v for k, v in opciones.items() if v is not None} while url: cuerpo = pedir(s, url, params).json() filas = cuerpo.get("value", []) # Corte por "value" vacio y no solo por falta de nextLink: cuando el # total es multiplo exacto del tamano de pagina, el servidor manda un # nextLink que apunta a una pagina vacia. if not filas: return for fila in filas: yield fila # el nextLink ya trae todos los parametros: no hay que re-enviarlos url, params = cuerpo.get("@odata.nextLink"), None def contar(s, entity_set, filtro=None): """Total de filas que coinciden con el filtro, sin traer los datos.""" params = {"$filter": filtro} if filtro else None return int(pedir(s, ROOT + entity_set + "/$count", params).text.strip()) # --------------------------------------------------------------------- # main # --------------------------------------------------------------------- def main(): if not API_KEY: raise SystemExit( "Falta la credencial.\n" " export SYNERGY_API_KEY=\"TU_API_KEY\"\n" " export SYNERGY_ODATA_ROOT=\"https://TU-HOST/synergy-server/jwt-odata-key/\"" ) s = sesion() print("Service root: %s\n" % ROOT) print("== Entidades disponibles") disponibles = entidades_disponibles(s) if not disponibles: raise SystemExit("El cliente no tiene ninguna entidad habilitada. " "Pedirle al administrador que las asigne.") for e in disponibles: print(" %-28s %s" % (e["nombre"], e["titulo"])) entity_set = sys.argv[1] if len(sys.argv) > 1 else disponibles[0]["nombre"] print("\n== Modelo de '%s'" % entity_set) modelo = describir(s, entity_set) if modelo is None: raise SystemExit("'%s' no esta en el $metadata. Nombres validos: %s" % (entity_set, ", ".join(e["nombre"] for e in disponibles))) print(" Etiqueta: %s" % (modelo["etiqueta"] or "(sin etiqueta)")) print(" Clave: %s" % ", ".join(modelo["claves"])) print(" Campos:") for p in modelo["propiedades"]: marca = "PK" if p["clave"] else (" " if p["nullable"] else "NN") largo = "(%s)" % p["largo"] if p["largo"] else "" print(" %s %-24s %-22s %s" % (marca, p["nombre"], p["tipo"] + largo, p["etiqueta"] or "")) if modelo["navegaciones"]: print(" Maestros relacionados:") for n in modelo["navegaciones"]: cardinalidad = "1 a N" if n["coleccion"] else "N a 1" print(" %-24s %s -> %s" % (n["nombre"], cardinalidad, n["destino"])) else: print(" Sin navegaciones publicadas para este cliente.") # una consulta de ejemplo armada con el modelo real, no con nombres fijos escalares = [p["nombre"] for p in modelo["propiedades"]][:4] una_nav = next((n for n in modelo["navegaciones"] if not n["coleccion"]), None) print("\n== Primeras filas") print(" Total: %d" % contar(s, entity_set)) opciones = {"select": ",".join(escalares), "top": "5"} if una_nav: opciones["expand"] = una_nav["nombre"] for i, fila in enumerate(recorrer(s, entity_set, **opciones), 1): valores = " ".join("%s=%r" % (c, fila.get(c)) for c in escalares) print(" %d. %s" % (i, valores)) if una_nav: relacionado = fila.get(una_nav["nombre"]) # None es legitimo: la FK puede estar en null print(" %s -> %s" % (una_nav["nombre"], relacionado)) if i >= 5: break print("\nListo. Para el detalle de $filter, $expand y las prestaciones no " "implementadas, ver api-docs/odata-api.html") if __name__ == "__main__": main()