The Problem
Enterprise clients have systems that predate REST, JSON, and sometimes HTTP itself. SOAP services, XML-RPC APIs, FTP-based data exchanges, and mainframe terminals are all still in production. Your AI system needs to integrate with these. This lesson covers SOAP-to-REST adapters, FTP polling, CSV/EDI parsing, and making old systems talk to modern AI pipelines.
Consuming SOAP services with zeep
# Zeep is a modern SOAP client for Python that handles WSDL parsing,
# XML namespace management, and SOAP envelope construction.
from zeep import Client, Settings
from zeep.transports import Transport
from zeep.wsse.username import UsernameToken
from requests.auth import HTTPBasicAuth, HTTPNTLMAuth
from requests import Session
import requests
# --- Basic SOAP client from WSDL ---
def create_soap_client(wsdl_url: str, username: str = None, password: str = None) -> Client:
"""Create a Zeep SOAP client from a WSDL URL with optional auth."""
session = Session()
# Basic auth
if username and password:
session.auth = HTTPBasicAuth(username, password)
transport = Transport(session=session, timeout=30, operation_timeout=60)
settings = Settings(strict=False, raw_response=False) # strict=False for non-standard WSDLs
return Client(wsdl_url, transport=transport, settings=settings)
# --- NTLM auth (Windows / IIS-hosted SOAP services) ---
def create_ntlm_soap_client(wsdl_url: str, username: str, password: str, domain: str) -> Client:
"""Create a SOAP client with NTLM authentication for IIS-hosted services."""
session = Session()
session.auth = HTTPNTLMAuth(username=f"{domain}\\{username}", password=password)
transport = Transport(session=session)
return Client(wsdl_url, transport=transport)
# --- WS-Security username token ---
def create_wsse_soap_client(wsdl_url: str, username: str, password: str) -> Client:
"""Create a SOAP client with WS-Security UsernameToken."""
wsse = UsernameToken(username, password)
return Client(wsdl_url, wsse=wsse)
# --- Calling a SOAP operation ---
def get_customer_from_sap(client: Client, customer_id: str) -> dict:
"""Call a SAP SOAP service to get customer details."""
# Inspect available operations:
# for service in client.wsdl.services:
# for port in service.ports:
# print(f"{service.name}.{port.name}: {port.binding.operations}")
# Call the operation
response = client.service.ZGetCustomerDetails(
CustomerID=customer_id,
# Zeep handles XML namespace wrapping automatically
)
# Zeep returns Python dicts from the SOAP response
return {
"customer_id": response["CustomerID"],
"name": response["Name"],
"city": response["City"],
"country": response["Country"],
}
XML to JSON transformation
import xml.etree.ElementTree as ET
import json
from collections import defaultdict
def xml_to_dict(element: ET.Element) -> dict:
"""Convert XML element to nested dict — handles attributes and repeated children."""
result = {}
# Add attributes with @ prefix
for key, value in element.attrib.items():
result[f"@{key}"] = value
# Add text content
if element.text and element.text.strip():
result["#text"] = element.text.strip()
# Handle child elements
children = defaultdict(list)
for child in element:
children[child.tag].append(xml_to_dict(child))
for tag, items in children.items():
if len(items) == 1:
result[tag] = items[0]
else:
result[tag] = items # Repeated elements become arrays
return result
def soap_response_to_json(soap_xml: str) -> dict:
"""Parse a SOAP response envelope and extract the payload as JSON."""
root = ET.fromstring(soap_xml)
# SOAP namespace
ns = {"soap": "http://schemas.xmlsoap.org/soap/envelope/"}
# Extract Body > payload
body = root.find(".//soap:Body", ns)
if body is None:
return {"error": "No SOAP Body found"}
# Get first child of Body (the actual response payload)
payload = list(body)[0] if len(body) > 0 else None
if payload is None:
return {"error": "Empty SOAP Body"}
return xml_to_dict(payload)
Legacy authentication: NTLM, kerberos, custom tokens
import requests
from requests_ntlm import HttpNtlmAuth
import base64
class LegacyAuthHandler:
"""Handle legacy authentication schemes found in enterprise systems."""
@staticmethod
def basic_auth(username: str, password: str) -> dict:
"""HTTP Basic Auth — still used by many SOAP services."""
credentials = base64.b64encode(f"{username}:{password}".encode()).decode()
return {"Authorization": f"Basic {credentials}"}
@staticmethod
def ntlm_auth(session: requests.Session, username: str, password: str, domain: str):
"""NTLM auth for IIS-hosted services."""
session.auth = HttpNtlmAuth(f"{domain}\\{username}", password)
return session
@staticmethod
def kerberos_auth(session: requests.Session):
"""Kerberos (SPNEGO) auth — requires a valid TGT via kinit."""
from requests_kerberos import HTTPKerberosAuth, OPTIONAL
session.auth = HTTPKerberosAuth(mutual_authentication=OPTIONAL)
return session
@staticmethod
def custom_token_auth(token: str, header_name: str = "X-Auth-Token") -> dict:
"""Custom token scheme — many legacy systems use proprietary headers."""
return {header_name: token}
Connecting to legacy databases via ODBC
import pyodbc
import pandas as pd
class LegacyDatabaseBridge:
"""Connect to legacy databases (SQL Server, Oracle, DB2) via ODBC."""
def __init__(self, connection_string: str):
self.connection_string = connection_string
def query_to_dataframe(self, query: str) -> pd.DataFrame:
"""Execute a SQL query and return results as a DataFrame."""
with pyodbc.connect(self.connection_string) as conn:
return pd.read_sql(query, conn)
def sync_table_to_modern_db(self, table_name: str, target_conn, batch_size: int = 1000):
"""Sync a legacy table to a modern database in batches."""
offset = 0
while True:
query = f"SELECT * FROM {table_name} ORDER BY id OFFSET {offset} ROWS FETCH NEXT {batch_size} ROWS ONLY"
df = self.query_to_dataframe(query)
if df.empty:
break
# Write to modern DB (e.g., Postgres)
df.to_sql(table_name, target_conn, if_exists="append", index=False)
offset += batch_size
print(f"Synced {offset} rows from {table_name}")
# Connection string examples:
# SQL Server: "DRIVER={ODBC Driver 17 for SQL Server};SERVER=legacy-sql;DATABASE=ERP;UID=user;PWD=pass"
# Oracle: "DRIVER={Oracle ODBC Driver};DBQ=oracle-prod:1521/PROD;UID=user;PWD=pass"
# DB2: "DRIVER={IBM DB2 ODBC DRIVER};DATABASE=MAINDB;HOSTNAME=db2-host;PORT=50000;PROTOCOL=TCPIP;UID=user;PWD=pass"
Building adapter layers: SOAP-to-REST
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
app = FastAPI()
class CustomerRequest(BaseModel):
customer_id: str
class CustomerResponse(BaseModel):
customer_id: str
name: str
city: str
country: str
# Initialize SOAP client at startup
soap_client = None
@app.on_event("startup")
def init_soap_client():
global soap_client
soap_client = create_soap_client(
wsdl_url="https://legacy-sap.company.com/sap/bc/srt/wsdl/zcustomer",
username=os.environ["SAP_USER"],
password=os.environ["SAP_PASS"],
)
@app.get("/api/customers/{customer_id}", response_model=CustomerResponse)
async def get_customer(customer_id: str):
"""REST endpoint that wraps a legacy SOAP service.
Modern AI services call this REST endpoint.
Internally, it calls the SOAP service and transforms the response.
"""
try:
result = get_customer_from_sap(soap_client, customer_id)
return CustomerResponse(**result)
except Exception as e:
raise HTTPException(status_code=502, detail=f"SOAP service error: {str(e)}")
@app.post("/api/customers", response_model=CustomerResponse)
async def create_customer(req: CustomerRequest):
"""POST endpoint that wraps a SOAP create operation."""
try:
# Call SOAP create operation
response = soap_client.service.ZCreateCustomer(
CustomerID=req.customer_id,
)
return CustomerResponse(
customer_id=response["CustomerID"],
name=response["Name"],
city=response["City"],
country=response["Country"],
)
except Exception as e:
raise HTTPException(status_code=502, detail=f"SOAP create failed: {str(e)}")