Auckland, New Zealand
+64 2239 10000
gareth.cheyne@gmail.com

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__()