Skip to content

Instantly share code, notes, and snippets.

@rmorlok
Created October 27, 2013 04:07
Show Gist options
  • Save rmorlok/7177882 to your computer and use it in GitHub Desktop.
Save rmorlok/7177882 to your computer and use it in GitHub Desktop.
Connecting to the Marketo SOAP API using Python and suds
Marketo has a [SOAP-based API](http://developers.marketo.com/documentation/soap) that allows you to interact with a lot of their data, and while there is a [Python library that supports it](https://github.com/segmentio/marketo-python), the library doesn't cover nearly all the methods supported in the API.
[suds](https://fedorahosted.org/suds/) is a Python SOAP library that will read WSDL and provide methods for calling into the service.
Here is an example of making a call to the getLead(...):
import hmac
import hashlib
import datetime
import time
from suds.client import Client
def _utc_offset(date, use_system_timezone):
if isinstance(date, datetime.datetime) and date.tzinfo is not None:
return _timedelta_to_seconds(date.dst() or date.utcoffset())
elif use_system_timezone:
if date.year < 1970:
# We use 1972 because 1970 doesn't have a leap day (feb 29)
t = time.mktime(date.replace(year=1972).timetuple())
else:
t = time.mktime(date.timetuple())
if time.localtime(t).tm_isdst: # pragma: no cover
return -time.altzone
else:
return -time.timezone
else:
return 0
def rfc3339(date, utc=False, use_system_timezone=True):
# Try to convert timestamp to datetime
try:
if use_system_timezone:
date = datetime.datetime.fromtimestamp(date)
else:
date = datetime.datetime.utcfromtimestamp(date)
except TypeError:
pass
if not isinstance(date, datetime.date):
raise TypeError('Expected timestamp or date object. Got %r.' %
type(date))
if not isinstance(date, datetime.datetime):
date = datetime.datetime(*date.timetuple()[:3])
utc_offset = _utc_offset(date, use_system_timezone)
if utc:
return _string(date + datetime.timedelta(seconds=utc_offset), 'Z')
else:
return _string(date, _timezone(utc_offset))
def _string(d, timezone):
return ('%04d-%02d-%02dT%02d:%02d:%02d%s' %
(d.year, d.month, d.day, d.hour, d.minute, d.second, timezone))
def sign(message, encryption_key):
digest = hmac.new(encryption_key, message, hashlib.sha1)
return digest.hexdigest().lower()
def set_header(client, user_id, encryption_key):
h = client.factory.create('AuthenticationHeaderInfo')
h.mktowsUserId = user_id
h.requestTimestamp = rfc3339(datetime.datetime.now())
h.requestSignature = sign(h.requestTimestamp + user_id, encryption_key)
client.set_options(soapheaders=h)
url = '<pointer to your api service here>?WSDL'
client = Client(url)
set_header(client, '<your username here>', '<your secret key here>')
leadKey = client.factory.create('LeadKey')
leadKeyRef = client.factory.create('LeadKeyRef')
leadKey.keyType = leadKeyRef.EMAIL
leadKey.keyValue = 'bob@dole.com'
print client.service.getLead(leadKey=leadKey)
Note that some of the above code is copied from/based on the aforementioned [marketo-python library](https://github.com/segmentio/marketo-python)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment