MSCRM :: Python WebApi
For a while now I have been playing around with this project, and although it is not complete I have found it very useful.
#Resource URI, ie you CRM Instance
RESOURCE_URI: https://{your production url}.crm6.dynamics.com
SANDBOX_RESOURCE_URI: https://{your sandbox url}.crm6.dynamics.com
API_VERSION: 9.1
# Username and Password to CRM Instance
XRM_USERNAME: {username}
XRM_PASSWORD: {password}
# Azure Tenant Auth Url
TENANT_AUTHORIZATION_URL: https://login.windows.net/{app guid}/oauth2/token/
# Azure ClientID and Secret for Dynamics Api
XRM_CLIENTID: xxxxxxxx
XRM_CLIENTSECRET: xxxxxxxx
See my code below.
import requests
import json
import yaml
from pandas.io.json import json_normalize
from datetime import datetime, timedelta
# Resource URI, ie you CRM Instance
RESOURCE_URI = ''
API_VERSION = ''
# Azure Token Created by Script
API_TOKEN = {'token': None, 'expire_on': None}
class WebApiException(Exception):
pass
def GetToken(config_file_location):
"""
Connect to the Azure Authorization URL and get a token to us the WebApi Calls.
"""
with open(config_file_location) as ymlfile:
cfg = yaml.load(ymlfile)
RESOURCE_URI = str(cfg['RESOURCE_URI'])
SANDBOX_RESOURCE_URI = str(cfg['SANDBOX_RESOURCE_URI'])
API_VERSION = str(cfg['API_VERSION'])
XRM_USERNAME = cfg['XRM_USERNAME']
XRM_PASSWORD = cfg['XRM_PASSWORD']
TENANT_AUTHORIZATION_URL = cfg['TENANT_AUTHORIZATION_URL']
XRM_CLIENTID = cfg['XRM_CLIENTID']
XRM_CLIENTSECRET = cfg['XRM_CLIENTSECRET']
def CheckTokenExpire(secs):
"""
Sets the DateTime of when the Token Expires using the stand python datetime format
"""
now = datetime.now()
expire = now + timedelta(0, int(secs))
return expire
if API_TOKEN['expire_on'] is None or API_TOKEN['expire_on'] > datetime.now():
data = {
'client_id': XRM_CLIENTID,
'client_secret': XRM_CLIENTSECRET,
'resource': RESOURCE_URI,
'username': XRM_USERNAME,
'password': XRM_PASSWORD,
'grant_type': 'password'
}
token_response = requests.post(TENANT_AUTHORIZATION_URL, data=data)
if token_response.status_code is 200:
API_TOKEN['token'] = token_response.json()['access_token']
API_TOKEN['expire_on'] = CheckTokenExpire(token_response.json()['expires_in'])
print('pyXRM :: New Token')
return RESOURCE_URI, SANDBOX_RESOURCE_URI, API_VERSION, (API_TOKEN['token'])
else:
print(':( Sorry you have a connection error, please review your pyXRM config file.')
print('=== Stack Trace - Start ===')
print(token_response.json()['error_description'])
print('=== Stack Trace - End ===')
print('Exiting Script Now...')
exit()
else:
print('pyDynamicsWebApi :: Old Token')
return RESOURCE_URI, SANDBOX_RESOURCE_URI, API_VERSION, (API_TOKEN['token'])
class WebApi(object):
"""
List of all the standard Web Api called based on the standardised calls listed on MS Dynamics Web Api Dev site
https://docs.microsoft.com/en-us/dynamics365/customer-engagement/developer/clientapi/reference/xrm-webapi
"""
def __init__(self, config_file_location='xrm_config.yaml'):
self._resource_uri, self._sandbox_resource_uri, self._api_version, self._token = GetToken(config_file_location)
self._user = None
self._headers = {
'OData-MaxVersion': '4.0',
'OData-Version': '4.0',
'Accept': 'application/json',
'Authorization': 'Bearer ' + self._token,
'Content-Type': 'application/json; charset=utf-8',
'MSCRMCallerID': self._user,
}
def __connection_test__(self, instance=None):
"""
Basis test that you have configured your yaml file, and your credentials works. Response should be OrganizationId, UserId, and BusinessUnitID
:return: json response
"""
resource_uri = WebApi.__check_instance__(self, instance)
response = requests.get(resource_uri + '/api/data/v' + self._api_version + '/WhoAmI', headers=self._headers)
if response.status_code is not 200:
print('pyDynamicsWebApi :: Connection Test Failed')
print(response.status_code)
else:
for key, value in response.json().items():
print(key, value)
return
def __check_instance__(self, instance=None):
if instance is 'sandbox':
resource_uri = self._sandbox_resource_uri
else:
resource_uri = self._resource_uri
return resource_uri
def RetrieveRecord(self, entityLogicalName=None, id=None, options=None, user=None, instance=None):
"""
Retrieve a single record from Dynamics CRM, you must supply that records GUID
:param entityLogicalName:
:param id:
:param options:
:param user:
:param instance: Either None (Production), or Sandbox
:return:
"""
if user is not None:
self._headers.update({'MSCRMCallerID': user})
resource_uri = WebApi.__check_instance__(self, instance)
response = requests.get(resource_uri + '/api/data/v' + self._api_version + '/' + entityLogicalName + '/' + id + '?' + options, headers=self._headers)
data = response.json()
return data
def RetrieveMultipleRecords(self, entity=None, options=None, maxpagesize=None, user=None, instance=None):
"""
Retrieve multiple records from Dynamics CRM, you must supply that a query
:param entity: Dynamics logical schema name.
:param options: Dynamics Query
:param maxpagesize: Dynamics response is 5000 records by default.
:param user: Supply a Dynamics user GUID to impersonate a different user.
:param instance: Either None (Production), or Sandbox
:return:
"""
if options is None:
options = '?'
if user is not None:
self._headers.update({'MSCRMCallerID': user})
if maxpagesize is not None:
self._headers.update({'Prefer': 'odata.maxpagesize=' + str(maxpagesize)})
resource_uri = WebApi.__check_instance__(self, instance)
response = requests.get(resource_uri + '/api/data/v' + self._api_version + '/' + entity + options, headers=self._headers)
if 'error' in response.json():
print('pyDynamicsWebApi :: Error with RetrieveMultipleRecords request')
print(response.json())
data = response.json()
i = 0
while '@odata.nextLink' in response.json():
i += 1
print('Page Number = ' + str(i))
_nextLink = response.json()['@odata.nextLink']
response = requests.get(_nextLink, headers=self._headers)
data['value'].extend(response.json()['value'])
if 'error' in response.json():
print('pyDynamicsWebApi :: Error with RetrieveMultipleRecords request')
print(response.json())
print('pyDynamicsWebApi :: Success RetrieveMultipleRecords request')
return data['value']
def CreateRecord(self, entity=None, data=None, user=None, instance=None):
"""
Create Record in Dynamics CRM.
:param entity: Dynamics logical schema name.
:param data: Dynamics Query
:param user: Supply a Dynamics user GUID to impersonate a different user.
:param instance: Either None (Production), or Sandbox
:return:
"""
if user is not None:
self._headers.update({'MSCRMCallerID': user})
data = json.dumps(data)
resource_uri = WebApi.__check_instance__(self, instance)
response = requests.post(resource_uri + '/api/data/v' + self._api_version + '/' + entity, data=data, headers=self._headers)
data = response.json()
if 'error' in data:
print('pyDynamicsWebApi :: Error with CreateRecord request')
print(data)
return ''
print('pyDynamicsWebApi :: Success CreateRecords request')
return data
def UpdateRecord(self, entity=None, guid=None, alternatekey=None, data=None, user=None, instance=None):
"""
Update a Dynamics Entity Record witheith the GUID or AlternateKey ie product(accountnumber='1234')
:param entity: Required, A Dynamics entity logical name.
:param guid: Required if not alternatekey if offered.
:param alternatekey: Required if no guid is offered.
:param data: Required, A list of fields and the values you want updated.
:param user: Optional, A Dynamics user id you may want to masquerade as.
:param instance: Either None (Production), or Sandbox
:return: Dynamics Response in Python List Type
"""
if guid is None:
key = alternatekey
else:
key = guid
if user is not None:
self._headers.update({'MSCRMCallerID': user})
data = json.dumps(data)
resource_uri = WebApi.__check_instance__(self, instance)
response = requests.patch(resource_uri + '/api/data/v' + self._api_version + '/' + entity + '(' + key + ')', data=data, headers=self._headers)
data = response.json()
if 'error' in data:
print(data)
return data
def UpsertRecord(self, entity=None, guid=None, alternatekey=None, data=None, user=None, instance=None):
"""
Update a Dynamics Entity Record witheith the GUID or AlternateKey ie product(accountnumber='1234')
:param entity: Required, A Dynamics entity logical name.
:param guid: Required if not alternatekey if offered.
:param alternatekey: Required if no guid is offered.
:param data: Required, A list of fields and the values you want updated.
:param user: Optional, A Dynamics user id you may want to masquerade as.
:param instance: Either None (Production), or Sandbox
:return: Dynamics Response in Python List Type
"""
if guid is None:
key = alternatekey
else:
key = guid
if user is not None:
self._headers.update({'MSCRMCallerID': user})
self._headers.update({'If-None-Match': '*'})
data = json.dumps(data)
resource_uri = WebApi.__check_instance__(self, instance)
response = requests.patch(resource_uri + '/api/data/v' + self._api_version + '/' + entity + '(' + key + ')', data=data, headers=self._headers)
data = response.json()
if 'error' in data:
print(data)
return data
def DeleteRecord(self, entity=None, guid=None, alternatekey=None, user=None, instance=None):
if guid is None:
key = alternatekey
else:
key = guid
resource_uri = WebApi.__check_instance__(self, instance)
response = requests.delete(resource_uri + '/api/data/v' + self._api_version + '/' + entity + '(' + key + ')', headers=self._headers)
data = response.json()
if 'error' in data:
print('pyDynamicsWebApi :: ' + data['error']['message'])
return False
return True
def isAvailableOffline(self):
pass
def execute(self):
pass
def executeMultiple(self):
pass
def StatusCode(code):
if code is 200:
return str(code) + ' :: Success: Result Found'
elif code is 201:
return str(code) + ' :: Success: Record Created/Updated.'
elif code is '500':
return str(code) + ' :: Failed: Internal Server Error.'
else:
print(code)
return
@staticmethod
def ConvertToDictWithIndex(index_key=str, data=list):
"""
Converts the response from Dynamics to a Dictionary where you can control what field is used as the index key.
:param index_key: What field would you like as the Dictionary Key?
:param data: The JSON formatted response from Dynamics.
:return: A Dictionary Object with your desired key.
"""
if index_key is None:
print('pyDynamicsWebApi :: error, you must supply a key')
return
d = {}
if 'value' in data:
data = data['value']
for entry in data:
d[entry[index_key]] = {}
d[entry[index_key]] = entry
return d
@staticmethod
def SaveToCSV(filename='', location='', timestamp=bool, data=object):
"""
Converts the response from Dynamics Query to a CSV file.
:param filename: What you want the file to be called, will default to datadump.csv?
:param location: The location where you want the file saved, will default to where the script is running.
:param timestamp: Boolean, is you want the fame name to have the current time stamp.
:param data: The response data object from Dynamics.
"""
if filename is '':
if timestamp is True:
exe_time = datetime.now()
filename = exe_time.strftime('%d-%m-%Y-%H-%M-%S') + '-datadump_' + '.csv'
else:
filename = 'datadump.csv'
else:
if timestamp is True:
exe_time = datetime.now()
filename = exe_time.strftime('%d-%m-%Y-%H-%M-%S') + '-' + filename + '.csv'
else:
filename = filename + '.csv'
df = json_normalize(data)
df.to_csv(filename)
return
if __name__ == '__main__':
WebApi(config_file_location="../xrm_config.yaml").__connection_test__()
Recent Comments