Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
self.authenticate()
if filterType is None: raise ValueError("Invalid argument: required argument filterType is none.")
if filterValues is None: raise ValueError("Invalid argument: required argument filter_values is none.")
args = {
'access_token': self.token,
'_method': 'GET'
}
filterValues = filterValues.split() if type(filterValues) is str else filterValues
data=[('filterValues',(',').join(filterValues)), ('filterType', filterType)]
if fields is not None:
data.append(('fields',fields))
if batchSize is not None:
data.append(('batchSize',batchSize))
result_list = []
while True:
result = HttpLib().post(self.host + "/rest/v1/salespersons.json", args, data, mode='nojsondumps')
if result is None: raise Exception("Empty Response")
if not result['success'] : raise MarketoException(result['errors'][0])
result_list.extend(result['result'])
if len(result['result']) == 0 or 'nextPageToken' not in result:
break
args['nextPageToken'] = result['nextPageToken']
return result_list
def get_daily_errors(self):
self.authenticate()
args = {
'access_token' : self.token
}
result = HttpLib().get(self.host + "/rest/v1/stats/errors.json", args)
if result is None: raise Exception("Empty Response")
if not result['success'] : raise MarketoException(result['errors'][0])
return result['result']
def associate_lead(self, id, cookie):
self.authenticate()
if id is None: raise ValueError("Invalid argument: required argument id is none.")
if cookie is None: raise ValueError("Invalid argument: required argument cookie is none.")
args = {
'access_token': self.token,
'id': id,
'cookie': cookie
}
result = HttpLib().post(self.host + "/rest/v1/leads/" + str(id) + "/associate.json", args)
if result is None: raise Exception("Empty Response")
if not result['success'] : raise MarketoException(result['errors'][0])
return result['success'] # there is no 'result' node returned in this call
if id is None: raise ValueError("Invalid argument: required argument id is none.")
if dynamicContentId is None: raise ValueError("Invalid argument: required argument dynamicContentId is none.")
if segment is None: raise ValueError("Invalid argument: required argument segment is none.")
if value is None: raise ValueError("Invalid argument: required argument value is none.")
if type is None: raise ValueError("Invalid argument: required argument type is none.")
args = {
'access_token' : self.token
}
if type == 'HTML':
value = value.encode('ascii', 'xmlcharrefreplace')
data = {
'segment': segment,
'value': value,
'type': type
}
result = HttpLib().post(self.host + "/rest/asset/v1/email/" + str(id) + "/dynamicContent/" +
str(dynamicContentId) + ".json", args, data, mode='nojsondumps')
if result is None: raise Exception("Empty Response")
if not result['success'] : raise MarketoException(result['errors'][0])
return result['result']
def get_email_content(self, id, status=None):
self.authenticate()
if id is None: raise ValueError("Invalid argument: required argument id is none.")
args = {
'access_token' : self.token
}
if status is not None:
args['status'] = status
result = HttpLib().get(self.host + "/rest/asset/v1/email/" + str(id) + "/content.json", args)
if result is None: raise Exception("Empty Response")
if not result['success'] : raise MarketoException(result['errors'][0])
return result['result']
def create_update_companies(self, input, action=None, dedupeBy=None):
self.authenticate()
if input is None: raise ValueError("Invalid argument: required argument input is none.")
args = {
'access_token' : self.token
}
data={
'input': []
}
for company in input:
data['input'].append(company)
if action is not None:
data['action'] = action
if dedupeBy is not None:
data['dedupeBy'] = dedupeBy
result = HttpLib().post(self.host + "/rest/v1/companies.json", args, data)
if not result['success']: raise MarketoException(result['errors'][0])
return result['result']
def get_snippet_content(self, id):
self.authenticate()
if id is None: raise ValueError("Invalid argument: required argument id is none.")
args = {
'access_token' : self.token
}
result = HttpLib().get(self.host + "/rest/asset/v1/snippet/" + str(id) + "/content.json", args)
if result is None: raise Exception("Empty Response")
if not result['success'] : raise MarketoException(result['errors'][0])
return result['result']
def get_email_by_id(self, id, status=None):
self.authenticate()
if id is None: raise ValueError("Invalid argument: required argument id is none.")
args = {
'access_token': self.token
}
if status is not None:
args['status'] = status
result = HttpLib().get(self.host + "/rest/asset/v1/email/" + str(id) + ".json", args)
if result is None: raise Exception("Empty Response")
if not result['success'] : raise MarketoException(result['errors'][0])
return result['result']
def get_email_templates(self, maxReturn=None, status=None):
self.authenticate()
args = {
'access_token': self.token
}
if maxReturn is not None:
args['maxReturn'] = maxReturn
else:
maxReturn = 20
if status is not None:
args['status'] = status
result_list = []
offset = 0
while True:
result = HttpLib().get(self.host + "/rest/asset/v1/emailTemplates.json", args)
if result is None: raise Exception("Empty Response")
if not result['success'] : raise MarketoException(result['errors'][0])
if 'result' in result:
if len(result['result']) < maxReturn:
result_list.extend(result['result'])
break
else:
break
result_list.extend(result['result'])
offset += maxReturn
args['offset'] = offset
return result_list
def get_daily_usage(self):
self.authenticate()
args = {
'access_token' : self.token
}
result = HttpLib().get(self.host + "/rest/v1/stats/usage.json", args)
if result is None: raise Exception("Empty Response")
if not result['success'] : raise MarketoException(result['errors'][0])
return result['result']