Skip to content

Instantly share code, notes, and snippets.

@ojmccall
Created September 16, 2021 03:49
Show Gist options
  • Save ojmccall/b75aa435e7cefdd7480dca979c596b0e to your computer and use it in GitHub Desktop.
Save ojmccall/b75aa435e7cefdd7480dca979c596b0e to your computer and use it in GitHub Desktop.
import xml.etree.ElementTree as ET
import requests
client_id =
client_secret =
baseURL =
SOAPEndpoint = "https://"+baseURL+".soap.marketingcloudapis.com/Service.asmx"
#compile XML
XMLString = ("""<?xml version="1.0" encoding="UTF-8"?>
<s:Envelope xmlns:s="http://www.w3.org/2003/05/soap-envelope"
xmlns:a="http://schemas.xmlsoap.org/ws/2004/08/addressing"
xmlns:u="http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-wssecurity-utility-1.0.xsd">
<s:Header>
<a:Action s:mustUnderstand="1">Create</a:Action>
<a:To s:mustUnderstand="1">"""+SOAPEndpoint+"""</a:To>
<fueloauth xmlns="http://exacttarget.com">"""+token+"""</fueloauth>
</s:Header><s:Body xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:xsd="http://www.w3.org/2001/XMLSchema">
<CreateRequest xmlns="http://exacttarget.com/wsdl/partnerAPI">
<Options/>
<Objects xsi:type="DataExtension">
<Name> """ + DEName +"""</Name>
<CustomerKey>""" + DEName + """</CustomerKey>"""+Folder+"""
<IsSendable>false</IsSendable>
<Fields>""" +Fields + """</Fields>
</Objects>
</CreateRequest>
</s:Body>
</s:Envelope>""")
XML = bytes(XMLString,'utf-8')
print("XML here: "+str(XML))
#get token
try:
url = "https://"+baseURL+".auth.marketingcloudapis.com/v2/Token"
data = {"grant_type":"client_credentials",
"client_id":client_id,
"client_secret":client_secret
}
r = requests.post(url, data=data)
print("posted")
body = json.loads(r.content)
token = body['access_token']
print(body)
print(">>>body above")
print(token)
TokenStatus = r.status_code
if TokenStatus != 200:
Failure = 1
StatusMessage = "token request invalid"
else:
Failure = 0
except Exception as e: print("Error from Token request: "+e)
#SOAP API POST REQUEST
try:
header = {"Content-Type": "text/xml" }
post = requests.post(SOAPEndpoint, data = XML ,headers=header)
print("posted")
body = post.content
except Exception as e: print("error from DE Build: "+e)
#break down data into blocks of 1k and post to async API for DE just created by key
XMLString = body.decode("utf-8")
root = ET.fromstring(XMLString)
for elem in root.iter():
#print(elem.tag)
if "StatusMessage" in elem.tag:
#print(elem.text)
StatusMessage = elem.text
if "StatusCode" in elem.tag:
#print(elem.text)
StatusCode = elem.text
if "RequestID" in elem.tag:
#print(elem.text)
RequestId = elem.text
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment