-
Notifications
You must be signed in to change notification settings - Fork 27
Description
Description
Spyne is not compatible with Python 3.12 and appears to be in maintenance mode. We need to migrate our SOAP services (primarily in NoteReaderService and Epic integration) to a more modern, maintained solution that works with newer Python versions.
Context
healthchain/service/soap/wsgi.py- Current WSGI application setup for Spynehealthchain/service/soap/epiccdsservice.py- Epic CDS service implementation using Spynehealthchain/service/soap/model/- SOAP data models including fault modelshealthchain/gateway/services/notereader.py- NoteReader adapter that interacts with Spyne services
Request/Response Models
healthchain/models/requests/cdarequest.py- CDA request modelhealthchain/models/responses/cdaresponse.py- CDA response model
FastAPI Integration
healthchain/gateway/core/base.py- Base classes for gateway adapters
Possible Implementation
Replace Spyne with a combination of FastAPI + lxml for SOAP services while maintaining compatibility with existing Epic SOAP clients.
- FastAPI for HTTP server/request handling
- lxml for XML/SOAP parsing and generation
- Static or template-based WSDL generation
- Maintaining all current adapter patterns
Implementation Approach
1. SOAP Request Handler in FastAPI
from fastapi import FastAPI, Request, Response
from lxml import etree
import base64
app = FastAPI()
@app.post("/notereader")
async def process_document(request: Request):
# Parse incoming XML
raw_xml = await request.body()
soap_envelope = etree.fromstring(raw_xml)
# Define namespaces (match Epic's exactly)
namespaces = {
'soap': 'http://schemas.xmlsoap.org/soap/envelope/',
'urn': 'urn:epic-com:Common.2013.Services'
}
# Extract document from SOAP request
try:
document_xml = soap_envelope.xpath('//urn:ProcessDocument/urn:Document/text()',
namespaces=namespaces)
# Create CdaRequest object (reusing existing model)
cda_request = CdaRequest(document=document_xml[0])
# Process using existing handler
handler = your_adapter._handlers["ProcessDocument"]
result = handler(cda_request)
response = your_adapter._process_result(result)
# Generate SOAP response
soap_response = generate_soap_response(response)
return Response(
content=soap_response,
media_type="application/soap+xml"
)
except Exception as e:
# Generate SOAP fault
soap_fault = generate_soap_fault(str(e))
return Response(
content=soap_fault,
media_type="application/soap+xml",
status_code=500
)2. SOAP Response Generation
def generate_soap_response(response: CdaResponse) -> str:
# Create XML with lxml
envelope = etree.Element("{http://schemas.xmlsoap.org/soap/envelope/}Envelope")
body = etree.SubElement(envelope, "{http://schemas.xmlsoap.org/soap/envelope/}Body")
if response.error:
# Create fault
fault = etree.SubElement(body, "{http://schemas.xmlsoap.org/soap/envelope/}Fault")
faultcode = etree.SubElement(fault, "faultcode")
faultcode.text = "soap:Server"
faultstring = etree.SubElement(fault, "faultstring")
faultstring.text = response.error
else:
# Create success response (match Epic format exactly)
process_response = etree.SubElement(
body,
"{urn:epic-com:Common.2013.Services}ProcessDocumentResponse"
)
result = etree.SubElement(
process_response,
"{urn:epic-com:Common.2013.Services}ProcessDocumentResult"
)
# Add document content
result.text = response.document
return etree.tostring(envelope, encoding="unicode", pretty_print=True)
def generate_soap_fault(error_message: str) -> str:
envelope = etree.Element("{http://schemas.xmlsoap.org/soap/envelope/}Envelope")
body = etree.SubElement(envelope, "{http://schemas.xmlsoap.org/soap/envelope/}Body")
fault = etree.SubElement(body, "{http://schemas.xmlsoap.org/soap/envelope/}Fault")
faultcode = etree.SubElement(fault, "faultcode")
faultcode.text = "soap:Server"
faultstring = etree.SubElement(fault, "faultstring")
faultstring.text = error_message
return etree.tostring(envelope, encoding="unicode", pretty_print=True)3. WSDL Management - File-based Approach with Jinja2
from fastapi import FastAPI, Request, UploadFile, File
from fastapi.responses import Response
from fastapi.templating import Jinja2Templates
app = FastAPI()
templates = Jinja2Templates(directory="templates")
# Option 1: Serve static WSDL
@app.get("/notereader")
async def serve_wsdl(request: Request):
if "wsdl" in request.query_params:
# Serve WSDL from file
with open("static/epic_service.wsdl", "r") as f:
wsdl_content = f.read()
return Response(content=wsdl_content, media_type="application/xml")
# Handle normal SOAP request...
# Option 2: Template-based WSDL
@app.get("/notereader")
async def serve_template_wsdl(request: Request):
if "wsdl" in request.query_params:
# Get service URL for WSDL
service_url = str(request.base_url)
# Render WSDL template with appropriate values
return templates.TemplateResponse(
"epic_wsdl.xml",
{
"request": request,
"service_url": service_url,
"service_name": "ICDSServices",
"namespace": "urn:epic-com:Common.2013.Services",
},
media_type="application/xml"
)
# Handle normal SOAP request...
# Administrative endpoint to update WSDL file
@app.post("/admin/upload-wsdl")
async def upload_wsdl(file: UploadFile = File(...)):
content = await file.read()
# Save the uploaded WSDL
with open("static/epic_service.wsdl", "wb") as f:
f.write(content)
return {"filename": file.filename, "status": "WSDL uploaded successfully"}4. Sample Jinja2 WSDL Template
Create a file at templates/epic_wsdl.xml:
<?xml version="1.0" encoding="UTF-8"?>
<wsdl:definitions
name="{{ service_name }}"
targetNamespace="{{ namespace }}"
xmlns:soap="http://schemas.xmlsoap.org/wsdl/soap/"
xmlns:wsdl="http://schemas.xmlsoap.org/wsdl/"
xmlns:xsd="http://www.w3.org/2001/XMLSchema"
xmlns:tns="{{ namespace }}">
<!-- Type definitions -->
<wsdl:types>
<xsd:schema targetNamespace="{{ namespace }}">
<!-- Define your types here -->
<xsd:element name="ProcessDocument">
<xsd:complexType>
<xsd:sequence>
<xsd:element name="Document" type="xsd:string"/>
</xsd:sequence>
</xsd:complexType>
</xsd:element>
<xsd:element name="ProcessDocumentResponse">
<xsd:complexType>
<xsd:sequence>
<xsd:element name="ProcessDocumentResult" type="xsd:string"/>
</xsd:sequence>
</xsd:complexType>
</xsd:element>
</xsd:schema>
</wsdl:types>
<!-- Message definitions -->
<wsdl:message name="ProcessDocumentSoapIn">
<wsdl:part name="parameters" element="tns:ProcessDocument"/>
</wsdl:message>
<wsdl:message name="ProcessDocumentSoapOut">
<wsdl:part name="parameters" element="tns:ProcessDocumentResponse"/>
</wsdl:message>
<!-- Port Type -->
<wsdl:portType name="ICDSServicesSoap">
<wsdl:operation name="ProcessDocument">
<wsdl:input message="tns:ProcessDocumentSoapIn"/>
<wsdl:output message="tns:ProcessDocumentSoapOut"/>
</wsdl:operation>
</wsdl:portType>
<!-- Binding -->
<wsdl:binding name="ICDSServicesSoap" type="tns:ICDSServicesSoap">
<soap:binding style="document" transport="http://schemas.xmlsoap.org/soap/http"/>
<wsdl:operation name="ProcessDocument">
<soap:operation soapAction="{{ namespace }}/ProcessDocument"/>
<wsdl:input>
<soap:body use="literal"/>
</wsdl:input>
<wsdl:output>
<soap:body use="literal"/>
</wsdl:output>
</wsdl:operation>
</wsdl:binding>
<!-- Service -->
<wsdl:service name="{{ service_name }}">
<wsdl:port name="ICDSServicesSoap" binding="tns:ICDSServicesSoap">
<soap:address location="{{ service_url }}"/>
</wsdl:port>
</wsdl:service>
</wsdl:definitions>Implementation Steps
-
Preparation Phase:
- Extract current WSDL from Spyne application
- Document exact XML structure for requests/responses
- Set up automated tests using captured real-world examples
-
Basic Implementation:
- Set up FastAPI endpoints for SOAP requests
- Implement XML parsing for requests using lxml
- Implement response generation matching current format
- Serve static WSDL file extracted from Spyne
-
Advanced Implementation:
- Move to template-based WSDL generation
- Implement SOAP fault handling for error cases
- Add administrative endpoints for WSDL management
-
Testing & Verification:
- Test with SoapUI or similar tool
- Verify all operations work correctly
- Compare responses with current implementation
Key Considerations
-
XML Namespaces: Epic systems will be very specific about namespaces. Match them exactly.
-
Base64 Encoding: Handle any base64 encoding/decoding happening with document transfer.
-
Error Handling: SOAP faults need to follow specific format for Epic compatibility.
-
Deployment Strategy: Consider a phased approach where both implementations run in parallel.
-
Advantages of this Approach:
- Full compatibility with Python 3.9 through 3.12+
- More maintainable, modern codebase
- Better performance (FastAPI/lxml are highly optimized)
- Easier to customize for specific needs
References
Metadata
Metadata
Assignees
Labels
Type
Projects
Status