SOAP 应用程序接口

在本示例中，您将创建一个 soapApi 蓝图，该蓝图将使用 Port'sAPI 和webhook functionality 的组合来引用 SOAP API 定义文件中的所有路径。

要将 API 路径引用到 Port，需要使用一个脚本，根据 webhook 配置发送有关 API 定义的信息。

先决条件

创建以下蓝图定义和 webhook 配置:

SOAP blueprint

{
  "identifier": "soapApi",
  "description": "This blueprint represents a SOAP API in our software catalog",
  "title": "SOAP API",
  "icon": "ApiDoc",
  "schema": {
    "properties": {
      "url": {
        "type": "string",
        "title": "API URL",
        "format": "url"
      },
      "attributes": {
        "type": "object",
        "title": "Attributes"
      },
      "parameters": {
        "type": "object",
        "title": "Parameters"
      },
      "response": {
        "type": "object",
        "title": "Response"
      }
    },
    "required": []
  },
  "mirrorProperties": {},
  "calculationProperties": {},
  "relations": {}
}

Package webhook configuration

{
  "identifier": "soapApiMapper",
  "title": "SOAP API Mapper",
  "description": "A webhook configuration to ingest SOAP API paths from a file",
  "icon": "ApiDoc",
  "mappings": [
    {
      "blueprint": "soapApi",
      "itemsToParse": ".body.paths",
      "entity": {
        "identifier": ".item.path_name",
        "title": ".item.path_name",
        "properties": {
          "url": ".item.url",
          "attributes": ".item.attributes",
          "parameters": ".item.parameters",
          "response": ".item.response"
        }
      }
    }
  ],
  "enabled": true,
  "security": {}
}

使用 Port 的 API 和 Python 脚本

下面的示例片段展示了如何使用 Python 将 Port 的 API 和 webhook 与现有的 Pipelines 集成:

Python script example

## Import the needed libraries
import requests
import os
import xml.dom.minidom

# The URL for the webhook endpoint provided by Port
WEBHOOK_URL = os.environ["WEBHOOK_URL"]

def add_entity_to_port(entity_object):
    """A function to create the passed entity in Port using the webhook URL

    Params
    --------------
    entity_object: dict
        The entity to add in your Port catalog

    Returns
    --------------
    response: dict
        The response object after calling the webhook
    """
    headers = {"Content-Type": "application/json"}
    response = requests.post(WEBHOOK_URL, json=entity_object, headers=headers)
    return response.json()


def ingest_soap_request_data(soap_file):
    """This function takes a soap request XML file and extracts the path information into a
    JSON array object

    Params
    --------------
    soap_file: str
        The path to the xml file relative to the project's root folder

    Returns
    --------------
    response: dict
        The JSON data extracted from the xml file
    """

    dom = xml.dom.minidom.parse(soap_file)

    # Get the root object
    root_element = dom.documentElement
    root_namespace = root_element.tagName.split(":")[0]

    # Extract data from the SOAP request body
    body_element = root_element.getElementsByTagName(root_namespace + ':Body')[0]
    api_url = body_element.getAttribute('xmlns')

    # Extract paths, attributes and parameters for each child element
    child_elements = [child for child in body_element.childNodes if child.nodeType == child.ELEMENT_NODE]
    soap_request_object = []
    for child_element in child_elements:
        path_name = child_element.localName
        self_url = child_element.getAttribute("xmlns")
        parameters = {}
        for parameter_element in child_element.getElementsByTagName('*'):
            parameter_name = parameter_element.localName
            parameter_value = parameter_element.firstChild.data if parameter_element.firstChild else None
            parameters[parameter_name] = parameter_value

        soap_request_object.append({"path_name": path_name, "attributes": {"self_url": self_url}, "parameters": parameters, "api_url": api_url})
    return soap_request_object


def ingest_soap_respone_data(soap_file):
    """This function takes a soap response XML file path and extracts the path information into a
    JSON object

    Params
    --------------
    soap_file: str
        The path to the xml file relative to the project's root folder

    Returns
    --------------
    response: dict
        The JSON data extracted from the xml file
    """

    dom = xml.dom.minidom.parse(soap_file)

    # Get the root object
    root_element = dom.documentElement
    root_namespace = root_element.tagName.split(":")[0]

    # Extract data from the SOAP response body
    body_element = root_element.getElementsByTagName(root_namespace + ':Body')[0]
    child_elements = [child for child in body_element.childNodes if child.nodeType == child.ELEMENT_NODE]
    soap_response_object = {}
    for child_element in child_elements:
        path_name = child_element.localName
        data = {}
        for data_element in child_element.getElementsByTagName('*'):
            data_name = data_element.localName
            if not data_name.endswith("Result"):

                data_value = data_element.firstChild.data if data_element.firstChild else None
                data[data_name] = data_value

        soap_response_object[path_name] = data

    return soap_response_object

# Replace the values in these variables with the paths leading to your SOAP API definition XML files
soap_request_file = "your_soap_request.xml"
soap_response_file = "your_soap_response.xml"

request_object = ingest_soap_request_data(soap_request_file)
response_object = ingest_soap_respone_data(soap_response_file)

## Map each SOAP Request to its response data
for index, element in enumerate(request_object):
    element["response"] = response_object.get(element["path_name"]+"Response", {})
    request_object[index].update(element)

port_entity_object = {
    "paths": request_object
}
webhook_response = add_entity_to_port(port_entity_object)
print(webhook_response)