Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
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/opportunities.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 update_email(self, id, name=None, description=None):
self.authenticate()
if id is None: raise ValueError("Invalid argument: required argument id is none.")
args = {
'access_token': self.token
}
if name is not None:
args['name'] = name
if description is not None:
args['description'] = description
result = HttpLib().post(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_program_by_tag_type(self, tagType, tagValue):
self.authenticate()
if tagType is None: raise ValueError("Invalid argument: required argument tagType is none.")
if tagValue is None: raise ValueError("Invalid argument: required argument tagValue is none.")
args = {
'access_token': self.token,
'tagType': tagType,
'tagValue': tagValue
}
result = HttpLib().get(self.host + "/rest/asset/v1/program/byTag.json", args)
if result is None: raise Exception("Empty Response")
if not result['success']: raise MarketoException(result['errors'][0])
return result['result']
def update_file_content(self, id, file):
self.authenticate()
if id is None: raise ValueError("Invalid argument: required argument id is none.")
if file is None: raise ValueError("Invalid argument: required argument file is none.")
args = {
'access_token' : self.token
}
result = HttpLib().post(self.host + "/rest/asset/v1/file/" + str(id) + "/content.json", args, files=file,
filename="file")
if result is None: raise Exception("Empty Response")
if not result['success'] : raise MarketoException(result['errors'][0])
return result['result']
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
'describe_opportunity_role': self.describe_opportunity_role,
'create_update_opportunities_roles': self.create_update_opportunities_roles,
'delete_opportunity_roles': self.delete_opportunity_roles,
'get_opportunity_roles': self.get_opportunity_roles,
'describe_company': self.describe_company,
'create_update_companies': self.create_update_companies,
'delete_companies': self.delete_companies,
'get_companies': self.get_companies,
'describe_sales_person': self.describe_sales_person,
'create_update_sales_persons': self.create_update_sales_persons,
'delete_sales_persons': self.delete_sales_persons,
'get_sales_persons': self.get_sales_persons
}
result = method_map[method](*args,**kargs)
self.API_CALLS_MADE += 1
except MarketoException as e:
'''
601 -> auth token not valid
602 -> auth token expired
'''
if e.code in ['601', '602']:
continue
else:
raise Exception({'message':e.message, 'code':e.code})
break
return result
def get_list_by_id(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/v1/lists/" + 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_tag_by_name(self, name):
self.authenticate()
if name is None: raise ValueError("Invalid argument: required argument name is none.")
args = {
'access_token': self.token,
'name': name
}
result = HttpLib().get(self.host + "/rest/asset/v1/tagType/byName.json", args)
if result is None: raise Exception("Empty Response")
if not result['success']: raise MarketoException(result['errors'][0])
return result['result']
def delete_snippet(self, id):
self.authenticate()
if id is None: raise ValueError("Invalid argument: required argument id is none.")
args = {
'access_token': self.token
}
result = HttpLib().post(self.host + "/rest/asset/v1/snippet/" + str(id) + "/delete.json", args)
if result is None: raise Exception("Empty Response")
if not result['success'] : raise MarketoException(result['errors'][0])
return result['result']
def describe_opportunity_role(self):
self.authenticate()
args = {
'access_token' : self.token
}
result = HttpLib().get(self.host + "/rest/v1/opportunities/roles/describe.json", args)
if result is None: raise Exception("Empty Response")
if not result['success'] : raise MarketoException(result['errors'][0])
return result['result']