Downloading files from Oracle UCM in Python

Oracle UCM(Universal Content Management) exposes its data through SOAP API. I have a requirement where in BICC( Business Intelligence Cloud Connector) publishes data based on a scheduled job to UCM from Oracle Cloud ERP. This data then needs to be moved or replicated to on premises or other data warehouses, in my case Snowflake. In this part I shall show you how this can be done in python to get files from UCM. These files are nothing but data generated by BICC in csv formats and compressed in zip format. In another post I will show how to import this data into Snowflake.

Python Version      : 3.6
Python Library Used : Zeep
Oracle UCM          : 

1. Install zeep

pip install zeep

2. Check the WSDL specification for all the services exposed by Oracle UCM. URL for Oracle UCM SOAP API is https://<host>:443/idcws/GenericSoapPort?WSDL. You can use below command to check the same.

python -mzeep https://<host>:443/idcws/GenericSoapPort?WSDL

3. Below is a truncated response from the above command. Check the Bindings, Service, Global types and Prefixes.

Prefixes:
xsd: http://www.w3.org/2001/XMLSchema
ns0: http://www.oracle.com/UCM
Global elements:
ns0:Field(xsd:string, name: xsd:anySimpleType)
ns0:GenericRequest(ns0:Generic)
ns0:GenericResponse(ns0:Generic)
Global types:
xsd:anyType
ns0:Container(Field: ns0:Field[], ResultSet: ns0:ResultSet[], OptionList: ns0:OptionList[], _attr_1: {})
ns0:File(Contents: xsd:base64Binary, name: xsd:anySimpleType, href: xsd:anySimpleType)
ns0:Generic(Service: ns0:Service, webKey: xsd:anySimpleType)
ns0:OptionList(Option: xsd:string[], name: xsd:anySimpleType)
ns0:ResultSet(Row: ns0:Row[], name: xsd:anySimpleType)
ns0:Row(Field: ns0:Field[], _attr_1: {})
ns0:Service(User: ns0:Container, Document: {Field: ns0:Field[], ResultSet: ns0:ResultSet[], OptionList: ns0:OptionList[], File: ns0:File[], _attr_1: {}}, IdcService: xsd:anySimpleType)
<Truncated>
Bindings:
Soap11Binding: {urn:GenericSoap}GenericSoapPortTypeSoapHttp
Service: GenericSoapService
Port: GenericSoapPort (Soap11Binding: {urn:GenericSoap}GenericSoapPortTypeSoapHttp)
Operations:
GenericSoapOperation(Service: ns0:Service, webKey: xsd:anySimpleType) -> Service: ns0:Service, webKey: xsd:anySimpleType

4. So the above API exposes one Service by name GenericSoapOperation. So all our calls are going to be through this.

GenericSoapOperation(Service: ns0:Service, webKey: xsd:anySimpleType) -> Service: ns0:Service, webKey: xsd:anySimpleType

5. The above syntax, tells us that this takes in Service and webKey as two mandatory input parameters.

6. Further looking at Global Types definition for Service, we have

ns0:Service(User: ns0:Container, Document: {Field: ns0:Field[], ResultSet: ns0:ResultSet[], OptionList: ns0:OptionList[], File: ns0:File[], _attr_1: {}}, IdcService: xsd:anySimpleType)

7. This syntax translates to an XML that is as below( example for downloading a file):

<ns1:GenericRequest webKey="cs">
<ns1:Service IdcService=<OPERATION>>
<ns1:User/>
<ns1:Document>
<ns1:Field name="dID"><DOCID></ns1:Field>
</ns1:Document>
</ns1:Service>
</ns1:GenericRequest>

Zeep generates this xml behind the scenes and encapsulates this in SOAP envelope.

8. Now lets start, coding.

# Import for SOAP Requests
from requests import Session
from requests.auth import HTTPBasicAuth
from zeep import Client
from zeep.transports import Transport
# To enable SOAP request/ response debugging. This will show what is the XML being sent/ received 
import logging.config
import zipfile
import os
# Lets create the client
session = Session()
session.auth = HTTPBasicAuth(<userName>,<passWord>)
client = Client('https://<host>:443/idcws/GenericSoapPort?WSDL',transport=Transport(session=session))
# Create Service Request
ser_request = client.get_type('ns0:Service')
ser = ser_request(Document= { 'Field': { 'name':'dID','_value_1':'371437'} } , IdcService= 'GET_FILE')
# Make request
res = client.service.GenericSoapOperation(Service=ser,webKey='cs')
# The response object will have the requested file as an attachment. Lets process this and save the file
fileName = <DIRECTORY TO WRITE TI> +res['Service']['Document']['File'][0].href
with open(fileName , 'wb') as f:
   f.write(res['Service']['Document']['File'][0].Contents)
if ( os.path.exists(fileName)) :
   zip = zipfile.ZipFile(fileName)
   zip.extractall()
# This will extract file by name res['Service']['Document']['File'][0].href to the current directory. If the file is not zipped then last part of extracting is not required.