Over the past several months, I've been doing a lot of work with OpenFaaS in my spare time, and in today's post I will show how you can use it to easily build and deploy a custom web service interface for data in a Dynamics 365 Customer Engagement online tenant.

OpenFaaS

If you're not familiar with OpenFaaS, it's basically a serverless functions platform like Azure Functions or AWS Lambda, but you run it on Kubernetes or Docker Swarm on your own servers or in the cloud. What I particularly like about OpenFaaS compared to the various commercial serverless platforms is that in addition to offering more control over how/where it's deployed, OpenFaaS supports a wider variety of languages for writing serverless functions.

OpenFaaS (Functions as a Service) is a framework for building serverless functions with Docker and Kubernetes which has first class support for metrics. Any process can be packaged as a function enabling you to consume a range of web events without repetitive boiler-plate coding.

To follow along with the samples in this post, you'll need access to a cluster with OpenFaaS deployed, so if you don't already have one, now would be an excellent time to look at the OpenFaaS deployment docs or maybe even work through the hands-on workshop. I've also previously written about how to securely deploy OpenFaaS on a free Google Cloud VM with Docker Swarm or on an Azure Kubernetes Service cluster.

Preparing to build the interface function

As soon as you have OpenFaaS running, it's time to look at the actual custom interface function.

My demo C# function does the following:

1. Parse a JSON object sent in the client request for an access key and optional query filter
2. Validate the client-supplied access key to authorize or reject the request
3. Retrieve a Dynamics 365 OAuth access token using my Azure AD OAuth 2 helper microservice
4. Execute a query for contacts against the Dynamics 365 Web API
5. Return the Web API query results to the client in an array as part of a JSON object

Because the OpenFaaS function uses my OAuth helper microservice instead of requesting an OAuth access token directly from Azure Active Directory, you need to deploy that microservice to your cluster before moving forward.

If you're using Kubernetes, you can create the deployment and corresponding service using the following YAML. You'll need to set the RESOURCE environment variable to the FQDN for your Dynamics 365 CE organization, but you can leave the CLIENTID and TOKEN_ENDPOINT values alone. (While I used to think you needed to register a separate client application for every Dynamics 365 org to use OAuth authentication, I recently learned via a Twitter conversation that there is a "universal" CRM client id you can use instead.)

apiVersion: apps/v1beta1
kind: Deployment
metadata:
  name: azuread-oauth2-helper
spec:
  replicas: 1
  template:
    metadata:
      labels:
        app: azuread-oauth2-helper
    spec:
      containers:
      - name: azuread-oauth2-helper
        image: lucasalexander/azuread-oauth2-helper
        ports:
        - containerPort: 5000
        env:
        - name: RESOURCE
          value: "https://XXXXXXXX.crm.dynamics.com"
        - name: CLIENTID
          value: "2ad88395-b77d-4561-9441-d0e40824f9bc"
        - name: TOKEN_ENDPOINT
          value: "https://login.microsoftonline.com/common/oauth2/token"
---
apiVersion: v1
kind: Service
metadata:
  name: azuread-oauth2-helper
spec:
  ports:
  - port: 5000
  selector:
    app: azuread-oauth2-helper

Once you've deployed the microservice, here's the definition for a Kubernetes ingress. In this case my microservice is accessible on the same host as OpenFaaS (akskube.alexanderdevelopment.net), and it is secured with the same Let's Encrypt certificate. You'll want to update your configuration with the appropriate values for your specific situation.

apiVersion: extensions/v1beta1
kind: Ingress
metadata:
  name: azuread-oauth2-helper-ingress
  annotations:
    kubernetes.io/tls-acme: "true"
    certmanager.k8s.io/issuer: letsencrypt-production
    nginx.ingress.kubernetes.io/rewrite-target: /
spec:
  tls:
  - hosts:
    - akskube.alexanderdevelopment.net
    secretName: faas-letsencrypt-production
  rules:
  - host: akskube.alexanderdevelopment.net
    http:
      paths:
      - path: /oauthhelper
        backend:
          serviceName: azuread-oauth2-helper
          servicePort: 5000

After the OAuth helper microservice is deployed, you should validate that you can get a token returned for a valid username/password combination. Here's what that looks like in Postman.

Building the interface function

If you've made it to this point, building and deploying the function is easy!

First the function gets its configuration data from environment variables that are set when the function is deployed. If you were actually using this function in production, it would be better to store sensitive values like the access key and the Dynamics 365 password as secrets, but I've used environment variables here to keep this overview as simple as possible.

//get configuration from env variables        
var username = Environment.GetEnvironmentVariable("USERNAME");
var userpassword = Environment.GetEnvironmentVariable("USERPASS");
var tokenendpoint = Environment.GetEnvironmentVariable("TOKENENDPOINT");
var accesskey = Environment.GetEnvironmentVariable("ACCESSKEY");
var crmwebapi = Environment.GetEnvironmentVariable("CRMAPI");

After the function gets its configuration data, it deserializes the client request using Json.Net to extract a client-supplied access key and an optional query filter. The client-supplied key is validated against the stored key value, and if they don't match, an error response is returned.

var queryrequest = JsonConvert.DeserializeObject<QueryRequest>(input);

if(accesskey!=queryrequest.AccessKey)
{
    JObject outputobject = new JObject();
    outputobject.Add("error", "Invalid access key");
    Console.WriteLine(outputobject.ToString());
    return;
}

After the access key is validated, the function then makes a request to the authentication helper microservice to get an access token.

var token = GetToken(username, userpassword, tokenendpoint);

...
...
...

string GetToken(string username, string userpassword, string tokenendpoint){
    try
    {
        JObject tokencredentials = new JObject();
        tokencredentials.Add("username", username);
        tokencredentials.Add("password",userpassword);
        var reqcontent = new StringContent(tokencredentials.ToString(), Encoding.UTF8, "application/json");
        var result = _client.PostAsync(tokenendpoint, reqcontent).Result;
        var tokenobj = JsonConvert.DeserializeObject<Newtonsoft.Json.Linq.JObject>(
            result.Content.ReadAsStringAsync().Result);
        var token = tokenobj["accesstoken"];
        return token.ToString();
    }
    catch(Exception ex)
    {
        return string.Format("Error: {0}", ex.Message);
    }
}

Once the token is returned from the microservice, the function executes the Web API query. The query is just a hardcoded OData query in the form of /contacts?$select=fullname,contactid plus any filter supplied by the client. The function expects that the filter will also be provided in supported Dynamics 365 OData format like startswith(fullname,'y').

var crmreq = new HttpRequestMessage(HttpMethod.Get, crmwebapi + crmwebapiquery);
crmreq.Headers.Add("Authorization", "Bearer " + token);
crmreq.Headers.Add("OData-MaxVersion", "4.0");
crmreq.Headers.Add("OData-Version", "4.0");
crmreq.Headers.Add("Prefer", "odata.maxpagesize=500");
crmreq.Headers.Add("Prefer", "odata.include-annotations=OData.Community.Display.V1.FormattedValue");
crmreq.Content = new StringContent(string.Empty.ToString(), Encoding.UTF8, "application/json");
var crmres = _client.SendAsync(crmreq).Result;

var crmresponse = crmres.Content.ReadAsStringAsync().Result;

var crmresponseobj = JsonConvert.DeserializeObject<Newtonsoft.Json.Linq.JObject>(crmresponse);

Finally results are returned to the client in an array as part of a JSON object.

JArray outputarray = new JArray();
foreach(var row in crmresponseobj["value"].Children())
{
    JObject record = new JObject();
    record.Add("id", row["contactid"]);
    record.Add("fullname", row["fullname"]);
    outputarray.Add(record);
}
JObject outputobject = new JObject();
outputobject.Add("contacts", outputarray);
Console.WriteLine(outputobject.ToString());

Here's the complete function.

using System;
using System.Text;
using System.Net;
using System.Net.Http;
using System.Net.Http.Headers;
using System.IO;
using Newtonsoft.Json;
using Newtonsoft.Json.Linq;
using System.Collections.Generic;

namespace Function
{
    public class FunctionHandler
    {
        private static HttpClient _client = new HttpClient();

        public void Handle(string input) {
            //get configuration from env variables        
            var username = Environment.GetEnvironmentVariable("USERNAME");
            var userpassword = Environment.GetEnvironmentVariable("USERPASS");
            var tokenendpoint = Environment.GetEnvironmentVariable("TOKENENDPOINT");
            var accesskey = Environment.GetEnvironmentVariable("ACCESSKEY");
            var crmwebapi = Environment.GetEnvironmentVariable("CRMAPI");
            
            //deserialize the client request
            var queryrequest = JsonConvert.DeserializeObject<QueryRequest>(input);
            
            //validate the client access key
            if(accesskey!=queryrequest.AccessKey)
            {
                JObject outputobject = new JObject();
                outputobject.Add("error", "Invalid access key");
                Console.WriteLine(outputobject.ToString());
                return;
            }

            //get the oauth token
            var token = GetToken(username, userpassword, tokenendpoint);
            
            if(!token.ToUpper().StartsWith("ERROR:"))
            {
                //set the base odata query
                var crmwebapiquery = "/contacts?$select=fullname,contactid";
                
                //add a filter if the client included one in the request
                if(!string.IsNullOrEmpty(queryrequest.Filter))
                    crmwebapiquery+="&$filter="+queryrequest.Filter;
                try
                {
                    //make the request to d365
                    var crmreq = new HttpRequestMessage(HttpMethod.Get, crmwebapi + crmwebapiquery);
                    crmreq.Headers.Add("Authorization", "Bearer " + token);
                    crmreq.Headers.Add("OData-MaxVersion", "4.0");
                    crmreq.Headers.Add("OData-Version", "4.0");
                    crmreq.Headers.Add("Prefer", "odata.maxpagesize=500");
                    crmreq.Headers.Add("Prefer", "odata.include-annotations=OData.Community.Display.V1.FormattedValue");
                    crmreq.Content = new StringContent(string.Empty.ToString(), Encoding.UTF8, "application/json");
                    var crmres = _client.SendAsync(crmreq).Result;
                    
                    //handle the d365 response
                    var crmresponse = crmres.Content.ReadAsStringAsync().Result;

                    var crmresponseobj = JsonConvert.DeserializeObject<Newtonsoft.Json.Linq.JObject>(crmresponse);
                    
                    try
                    {
                        //build the function response
                        JArray outputarray = new JArray();
                        foreach(var row in crmresponseobj["value"].Children())
                        {
                            JObject record = new JObject();
                            record.Add("id", row["contactid"]);
                            record.Add("fullname", row["fullname"]);
                            outputarray.Add(record);
                        }
                        JObject outputobject = new JObject();
                        outputobject.Add("contacts", outputarray);
                        
                        //return the response to the client
                        Console.WriteLine(outputobject.ToString());
                    }
                    catch(Exception ex)
                    {
                        JObject outputobject = new JObject();
                        outputobject.Add("error", string.Format("Could not parse query response: {0}", ex.Message));
                        Console.WriteLine(outputobject.ToString());
                    }
                }
                catch(Exception ex)
                {
                    JObject outputobject = new JObject();
                    outputobject.Add("error", string.Format("Could not query data: {0}", ex.Message));
                    Console.WriteLine(outputobject.ToString());
                }
            }
            else
            {
                JObject outputobject = new JObject();
                outputobject.Add("error", "Could not get token");
                Console.WriteLine(outputobject.ToString());
            }
        }

        string GetToken(string username, string userpassword, string tokenendpoint){
            try
            {
                JObject tokencredentials = new JObject();
                tokencredentials.Add("username", username);
                tokencredentials.Add("password",userpassword);
                var reqcontent = new StringContent(tokencredentials.ToString(), Encoding.UTF8, "application/json");
                var result = _client.PostAsync(tokenendpoint, reqcontent).Result;
                var tokenobj = JsonConvert.DeserializeObject<Newtonsoft.Json.Linq.JObject>(
                    result.Content.ReadAsStringAsync().Result);
                var token = tokenobj["accesstoken"];
                return token.ToString();
            }
            catch(Exception ex)
            {
                return string.Format("Error: {0}", ex.Message);
            }
        }
    }

    public class QueryRequest
    {
        public string AccessKey {get;set;}
        public string Filter{get;set;}
    }
}

Because the function relies on Json.Net, you need to add a reference to it in your .csproj file before you build the function.

<Project Sdk="Microsoft.NET.Sdk">
  <PropertyGroup>
    <TargetFramework>netstandard2.0</TargetFramework>
  </PropertyGroup>
  <PropertyGroup>
    <GenerateAssemblyInfo>false</GenerateAssemblyInfo>
  </PropertyGroup>
  <ItemGroup>
    <PackageReference Include="newtonsoft.json" Version="11.0.2" />
  </ItemGroup>
</Project>

Here is my function definition YAML file with enviroment variables included. You will need to update them with your appropriate values, and you will also need to change the image name if you're building your own function instead of just deploying mine from Docker Hub.

provider:
  name: faas
  gateway: http://localhost:8080

functions:
  demo-crm-function:
    lang: csharp
    handler: ./demo-crm-function
    image: lucasalexander/faas-demo-crm-function
    environment:
      USERNAME: XXXXXX@XXXXXX.onmicrosoft.com
      USERPASS: XXXXXX
      TOKENENDPOINT: https://akskube.alexanderdevelopment.net/oauthhelper/requesttoken
      CRMAPI: https://lucastest20.api.crm.dynamics.com/api/data/v9.0
      ACCESSKEY: MYACCESSKEY

Once the function is deployed, you can execute it either through the OpenFaaS admin UI or another tool that makes HTTP requests like Curl or Postman. Here's what an unfiltered query in Postman looks like for a Dynamics 365 org with sample data installed.

And here's a query with a filter included.

Wrapping up

Once I got OpenFaaS running, writing and deploying the actual function only took about an hour. Obviously writing a more complex data interface to support real-world requirements would take longer, but using a serverless functions platform like OpenFaaS is definitely a significant accelerator for custom Dynamics 365 integration development.

What do you think about this approach? Are you using serverless functions with your Dynamics 365 projects? What do you think about OpenFaaS vs Azure Functions or AWS Lambda? Let us know in the comments!

I was recently asked to be a guest on the third-anniversary episode of the CRM Audio podcast. While I was there George Doubinski challenged me to create a plugin in one Dynamics 365 organization to retrieve records from another Dynamics 365 organization so they could be displayed as virtual entities. I was promised adulation on Dynamics CRM Tip of the Day and fame beyond my wildest dreams, so naturally I accepted.

To address the challenge, I wrote a simple Dynamics 365 plugin that calls the Web API in a different Dynamics 365 organization to retrieve records and return them to a virtual entity data provider. From there, configuration of the Dynamics 365 virtual entity is simple. Let's take a look at how I did it.

The plugin

First you need to create a plugin to retrieve the data from the "external" Dynamics 365 org. Because this code connects directly to the Web API, you'll need to get an access token from Azure AD before you can make the request to Dynamics 365. Just like I showed in my "Scheduling Dynamics 365 workflows with Azure Functions and C#" post back in 2016, my sample code does not use ADAL to get the access token, but rather it issues a request directly to the Azure AD OAuth 2 token endpoint.

Here's the code for the plugin. There are some configuration values you'll need to set for your Dynamics 365 organization and whatever query you want to run. It's not a best practice to have any of this actually hardcoded in your plugin, but I've done it this way so it's easier to see how things work.

using Microsoft.Xrm.Sdk;
using Microsoft.Xrm.Sdk.Data.Exceptions;
using Microsoft.Xrm.Sdk.Extensions;
using Microsoft.Xrm.Sdk.Query;
using System;
using System.IO;
using System.Net;
using System.Threading.Tasks;
using Newtonsoft.Json;

namespace VirtualEntityProvider
{
    public class RetrieveOtherOrgData : IPlugin
    {
        //set these values for your D365 instance, user credentials and Azure AD clientid/token endpoint
        string crmorg = "https://XXXXX.crm.dynamics.com";
        string clientid = "XXXXXXXXX";
        string username = "lucasalexander@XXXXXX.onmicrosoft.com";
        string userpassword = "XXXXXXXXXXXX";
        string tokenendpoint = "https://login.microsoftonline.com/XXXXXXXXXXX/oauth2/token";

        //relative path to web api endpoint
        string crmwebapi = "/api/data/v8.2";

        //web api query to execute - in this case all accounts that start with "F"
        string crmwebapipath = "/accounts?$select=name,accountid&$filter=startswith(name,'F')";

        public void Execute(IServiceProvider serviceProvider)
        {
            //basic plugin set-up stuff
            IPluginExecutionContext context = (IPluginExecutionContext)serviceProvider.GetService(typeof(IPluginExecutionContext));
            IOrganizationServiceFactory servicefactory = (IOrganizationServiceFactory)serviceProvider.GetService(typeof(IOrganizationServiceFactory));
            IOrganizationService service = servicefactory.CreateOrganizationService(context.UserId);
            ITracingService tracingService = (ITracingService)serviceProvider.GetService(typeof(ITracingService));

            try
            {
                //instantiate a new entity collection to hold the records we'll return later
                EntityCollection results = new EntityCollection();

                //build the authorization request for Azure AD
                var reqstring = "client_id=" + clientid;
                reqstring += "&resource=" + Uri.EscapeUriString(crmorg);
                reqstring += "&username=" + Uri.EscapeUriString(username);
                reqstring += "&password=" + Uri.EscapeUriString(userpassword);
                reqstring += "&grant_type=password";

                //make the Azure AD authentication request
                WebRequest req = WebRequest.Create(tokenendpoint);
                req.ContentType = "application/x-www-form-urlencoded";
                req.Method = "POST";
                byte[] bytes = System.Text.Encoding.ASCII.GetBytes(reqstring);
                req.ContentLength = bytes.Length;
                System.IO.Stream os = req.GetRequestStream();
                os.Write(bytes, 0, bytes.Length);
                os.Close();

                HttpWebResponse resp = (HttpWebResponse)req.GetResponse();
                StreamReader tokenreader = new StreamReader(resp.GetResponseStream());
                string responseBody = tokenreader.ReadToEnd();
                tokenreader.Close();

                //deserialize the Azure AD token response and get the access token to supply with the web api query
                var tokenresponse = JsonConvert.DeserializeObject<Newtonsoft.Json.Linq.JObject>(responseBody);
                var token = tokenresponse["access_token"];

                //make the web api query
                WebRequest crmreq = WebRequest.Create(crmorg+crmwebapi+crmwebapipath);
                crmreq.Headers = new WebHeaderCollection();

                //use the access token from earlier as the authorization header bearer value
                crmreq.Headers.Add("Authorization", "Bearer " + token);
                crmreq.Headers.Add("OData-MaxVersion", "4.0");
                crmreq.Headers.Add("OData-Version", "4.0");
                crmreq.Headers.Add("Prefer", "odata.maxpagesize=500");
                crmreq.Headers.Add("Prefer", "odata.include-annotations=OData.Community.Display.V1.FormattedValue");
                crmreq.ContentType = "application/json; charset=utf-8";
                crmreq.Method = "GET";

                HttpWebResponse crmresp = (HttpWebResponse)crmreq.GetResponse();
                StreamReader crmreader = new StreamReader(crmresp.GetResponseStream());
                string crmresponseBody = crmreader.ReadToEnd();
                crmreader.Close();

                //deserialize the response
                var crmresponseobj = JsonConvert.DeserializeObject<Newtonsoft.Json.Linq.JObject>(crmresponseBody);

                //loop through the response values
                foreach (var row in crmresponseobj["value"].Children())
                {
                    //create a new virtual entity of type lpa_demove
                    Entity verow = new Entity("lpa_otheraccount");
                    //verow["lpa_otheraccountid"] = Guid.NewGuid();
                    //verow["lpa_name"] = ((Newtonsoft.Json.Linq.JValue)row["name"]).Value.ToString();
                    verow["lpa_otheraccountid"] = (Guid)row["accountid"];
                    verow["lpa_name"] = (string)row["name"];

                    //add it to the collection
                    results.Entities.Add(verow);
                }

                //return the results
                context.OutputParameters["BusinessEntityCollection"] = results;
            }
            catch (Exception e)
            {
                tracingService.Trace($"{e.Message} {e.StackTrace}");
                if (e.InnerException != null)
                    tracingService.Trace($"{e.InnerException.Message} {e.InnerException.StackTrace}");

                throw new InvalidPluginExecutionException(e.Message);
            }
        }
    }
}

Because the plugin uses JSON.Net, you'll need to use ILMerge to bundle the Newtonsoft.Json.dll assembly with your compiled plugin before you deploy it to Dynamics 365.

Setting up the virtual entity

After you've deployed the plugin using the plugin registration tool, register a new data provider. When the data provider registration window opens, first create a new data source entity.

Complete the details for the data source and save it.

Complete the rest of the details for the data provider and save it.

You should now see a new data provider and data source.

Open the Dynamics 365 web UI, and go to settings->administration->virtual entity data sources.

Click the "new" button to create a new virtual entity data source.

In the window that pops up, select the data provider you created earlier.

Give your new virtual entity data source a name and save it.

Open your solution and create a new entity.

Configure your entity as a virtual entity that uses the virtual entity data source you created previously.

Once you save and publish the virtual entity, you can open an advanced find view that will retrieve data from your other Dynamics 365 organization and display it.

If you export this data to Excel and unhide the id column, you will see that the GUIDs match the records in the external system.

And that's all there is to it. Happy entity virtualizing!

One of the biggest trends in systems architecture these days is the use of "serverless" functions like Azure Functions, Amazon Lambda and OpenFaas. Because these functions are stateless, if you want to use a purely serverless approach to work with resources secured using Azure Active Directory like Dynamics 365 online, a new token will have to be requested every time a function executes. This is inefficient, and it requires the function to fully understand OAuth 2 authentication, which could be handled better elsewhere.

To address this problem, I've written a microservice in Python that can be used to request OAuth 2 tokens from Azure Active Directory, and it also handles refreshing them as needed. I've containerized it as Docker image so you can easily run it without needing to build anything.

How it works

When a request containing a username and password arrives for the first time, the microservice retrieves an OAuth2 access token from Azure AD and returns it to the requester. The microservice also caches an object that contains the access token, refresh token, username, password and expiration time.

When subsequent requests arrive, the microservice checks its cache for an existing token that matches the username and password. If it finds one, it checks if the token has expired or needs to be refreshed.

If the existing token has expired, a new one is requested. If the existing token has not expired, but it will expire within a specified period of time (10 minutes is the default value), the microservice will execute a refresh request to Azure AD, cache the updated token and return it to the requester. If there's an unexpired existing token that doesn't need to be refreshed, the cached access token will be returned to the requester.

Here's what a raw token request to and response from the microservice looks like in Postman:

Back in 2016 I shared some sample Python code that showed how to authenticate to Azure AD and query the Dynamics 365 (then called Dynamics CRM) Web API. Here is an updated version of that sample code that uses this new microservice to acquire access tokens:

import requests
import json

#set these values to retrieve the oauth token
username = 'lucasalexander@xxxxxx.onmicrosoft.com'
userpassword = 'xxxxxx'
tokenendpoint = 'http://localhost:5000/requesttoken'

#set these values to query your crm data
crmwebapi = 'https://xxxxxx.api.crm.dynamics.com/api/data/v8.2'
crmwebapiquery = '/contacts?$select=fullname,contactid'

#build the authorization request
tokenpost = {
    'username':username,
    'password':userpassword
}

#make the token request
print('requesting token . . .')
tokenres = requests.post(tokenendpoint, json=tokenpost)
print('token response received. . .')

accesstoken = ''

#extract the access token
try:
    print('parsing token response . . .')
    print(tokenres)
    accesstoken = tokenres.json()['accesstoken']

except(KeyError):
    print('Could not get access token')

if(accesstoken!=''):
    crmrequestheaders = {
        'Authorization': 'Bearer ' + accesstoken,
        'OData-MaxVersion': '4.0',
        'OData-Version': '4.0',
        'Accept': 'application/json',
        'Content-Type': 'application/json; charset=utf-8',
        'Prefer': 'odata.maxpagesize=500',
        'Prefer': 'odata.include-annotations=OData.Community.Display.V1.FormattedValue'
    }

    print('making crm request . . .')
    crmres = requests.get(crmwebapi+crmwebapiquery, headers=crmrequestheaders)
    print('crm response received . . .')
    try:
        print('parsing crm response . . .')
        crmresults = crmres.json()
        for x in crmresults['value']:
            print (x['fullname'] + ' - ' + x['contactid'])
    except KeyError:
        print('Could not parse CRM results')

Here's the output when I run the sample against my demo environment:

Running the microservice

Pull the image from Docker Hub: docker pull lucasalexander/azuread-oauth2-helper:latest

Required environment variables

1. RESOURCE - The URL of the service that is going to be accessed
2. CLIENTID - The Azure AD application client ID
3. TOKEN_ENDPOINT - The OAuth2 token endpoint from the Azure AD application

Run the image with the following command (replacing the environment variables with your own).

docker run -d -p 5000:5000 -e RESOURCE=https://XXXXXX.crm.dynamics.com -e CLIENTID=XXXXXX -e TOKEN_ENDPOINT=https://login.microsoftonline.com/XXXXXX/oauth2/token --name oauthhelper lucasalexander/azuread-oauth2-helper:latest

You can also optionally supply an additional "REFRESH_THRESHOLD" environment variable that sets the time remaining (in seconds) before a token's expiration time when it will be refreshed. The default value is 600 seconds.

A note on security

Because the microservice is caching usernames, passwords and access tokens in memory, this approach is vulnerable to heap inspection attacks, so you'll want to make sure your environment is appropriately locked down. Also you'll want to make sure any communication between your code that requests tokens and the microservice is encrypted.

Wrapping up

Although I wrote this with Dynamics 365 in mind, it should work for any resource that is secured by Azure AD. If you'd like to take a closer look at the code, it's available on GitHub here.

I've released an updated version of my recurring workflow scheduler for Dynamics 365 Customer Engagement. This solution targets Dynamics 365 version 9, so it should work in all current Dynamics 365 online organizations. You can download version 1.3 of my solution from here: https://github.com/lucasalexander/AlexanderDevelopment.ProcessRunner/releases/tag/v1.3.

For more information on the use of this tool, take a look at the original blog posts:

• https://alexanderdevelopment.net/post/2016/09/19/updated-solution-for-scheduling-recurring-dynamics-crm-workflows/
• https://alexanderdevelopment.net/post/2013/05/18/scheduling-recurring-dynamics-crm-workflows-with-fetchxml/

Recently I was asked to set up a process to automatically disable or re-enable Dynamics 365 Customer Engagement users depending on some external data. This ended up being ridiculously easy to do with SSIS and KingswaySoft's Dynamics 365 Integration Toolkit. Let me show you how it works.

In Dynamics 365 CE, you can disable or enable a user record just by setting the value of its "isdisabled" attribute to true or false, so both my disable user data flow and re-enable user data flow do roughly the same thing.

1. Get a list of Dynamics 365 user records to update.
2. Add a derived column to hold the value to use for updating isdisabled on the user records.
3. Update the user records.

The disable users package

Here's a screenshot of a sample disable users data flow.

Let's take a closer look at each step.

1. Query users using FetchXML.
2. Add a derived column named "isdisabled" and set its value to 1.
3. Update the users.

Enabling the users works exactly the same way, except the value of the "isdisabled" column should be 0 instead of 1, so I won't show the screenshots for that package.

Disable user demo

In my Dynamics 365 online instance, I have an active user named "Angus Alexander" who I want to disable.

When I run the disable users package with the query from above (<condition attribute="firstname" operator="eq" value="angus" />) in Visual Studio, I see success on every step.

I check back in Dynamics 365 to see Angus Alexander is no longer an enabled user.

Instead Angus shows up as a disabled user.

Now when Angus tries to access Dynamics 365, he sees that his account has been disabled.

This is the final post in my series about building a service relay for Dynamics 365 CE with RabbitMQ and Python. In my previous post in this series, I showed the Python code to make the service relay work. In today's post, I will show how you can use Azure Functions to make a consumer service proxy using C# so client applications don't have to access to your RabbitMQ broker directly, and I will also discuss some general thoughts on security and scalability for this service relay architecture.

Although this simple service relay allows external consumers to get data from Dynamics 365 CE without needing to connect directly, the examples I've shown so far require that they can connect to a RabbitMQ broker. This may be problematic for a variety of reasons, so you would probably want external consumers to connect to a web service proxy that would write requests to and read responses from the RabbitMQ broker.

Building a service proxy function

You can build an Azure Functions service proxy with Python, but I don't recommend it for three reasons:

1. Azure Functions Python support is still considered experimental.
2. Python scripts that use external libraries can run exceedingly slow.
3. Getting the environment set up is a bit of a hassle.

On the other hand, building a service proxy function with C# was so much easier, and it performed much better than a comparable Python function (~.5 seconds for C# compared to 5+ seconds for Python).

Here are the steps I took to build my C# service proxy function:

1. Create a C# HTTP trigger function.
2. Create and upload a project.json file with a dependency on the RabbitMQ client (see below).
3. Take the "RpcClient" class from the RabbitMQ .Net RPC tutorial and call it from within my function.

Here's my project.json file:

{
  "frameworks": {
    "net46":{
      "dependencies": {
        "RabbitMQ.Client": "5.0.1"
      }
    }
   }
}

And here's my run.csx file:

using System.Net;
using System;
using System.Collections.Concurrent;
using System.Text;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;

public static async Task<HttpResponseMessage> Run(HttpRequestMessage req, TraceWriter log)
{
    log.Info("Processing request");

    // parse query parameter
    string query = req.GetQueryNameValuePairs()
        .FirstOrDefault(q => string.Compare(q.Key, "query", true) == 0)
        .Value;

    // Get request body
    dynamic data = await req.Content.ReadAsAsync<object>();

    // Set name to query string or body data
    query = query ?? data?.query;

    var rpcClient = new RpcClient();
    
    log.Info(string.Format(" [.] query start time {0}", DateTime.Now.ToString("MM/dd/yyyy hh:mm:ss.fff tt")));
    var response = rpcClient.Call(query);

    log.Info(string.Format(" [.] query end time {0}", DateTime.Now.ToString("MM/dd/yyyy hh:mm:ss.fff tt")));
    rpcClient.Close();

    return req.CreateResponse(HttpStatusCode.OK, response);
}

public class RpcClient
{
    private readonly IConnection connection;
    private readonly IModel channel;
    private readonly string replyQueueName;
    private readonly EventingBasicConsumer consumer;
    private readonly BlockingCollection<string> respQueue = new BlockingCollection<string>();
    private readonly IBasicProperties props;

    public RpcClient()
    {
        var factory = new ConnectionFactory() { HostName = "RABBITHOST", UserName="RABBITUSER", Password="RABBITUSERPASS"  };

        connection = factory.CreateConnection();
        channel = connection.CreateModel();
        replyQueueName = channel.QueueDeclare().QueueName;
        consumer = new EventingBasicConsumer(channel);

        props = channel.CreateBasicProperties();
        var correlationId = Guid.NewGuid().ToString();
        props.CorrelationId = correlationId;
        props.ReplyTo = replyQueueName;

        consumer.Received += (model, ea) =>
        {
            var body = ea.Body;
            var response = Encoding.UTF8.GetString(body);
            if (ea.BasicProperties.CorrelationId == correlationId)
            {
                respQueue.Add(response);
            }
        };
    }

    public string Call(string message)
    {
        var messageBytes = Encoding.UTF8.GetBytes(message);
        channel.BasicPublish(
            exchange: "",
            routingKey: "rpc_queue",
            basicProperties: props,
            body: messageBytes);

        channel.BasicConsume(
            consumer: consumer,
            queue: replyQueueName,
            autoAck: true);

        return respQueue.Take(); ;
    }

    public void Close()
    {
        connection.Close();
    }
}

Here's a screenshot showing me calling the C# function with Postman.

Because I did actually build a Python function, I will go ahead and share how I did it if you're interested. Here are the steps I took:

1. Create a Python HTTP trigger function.
2. Install Python 3.6 via site extensions (see steps 2.1-2.4 here).
3. Install the necessary libraries using pip via KUDU.

Here's the Python function code:

import os
import sys
import json
import pika
import uuid
import datetime

class CrmRpcClient(object):
    def __init__(self):
        #RabbitMQ connection details
        self.rabbituser = 'RABBITUSERNAME'
        self.rabbitpass = 'RABBITUSERPASS'
        self.rabbithost = 'RABBITHOST' 
        self.rabbitport = 5672
        self.rabbitqueue = 'rpc_queue'
        rabbitcredentials = pika.PlainCredentials(self.rabbituser, self.rabbitpass)
        rabbitparameters = pika.ConnectionParameters(host=self.rabbithost,
                                    port=self.rabbitport,
                                    virtual_host='/',
                                    credentials=rabbitcredentials)

        self.rabbitconn = pika.BlockingConnection(rabbitparameters)

        self.channel = self.rabbitconn.channel()

        #create an anonymous exclusive callback queue
        result = self.channel.queue_declare(exclusive=True)
        self.callback_queue = result.method.queue

        self.channel.basic_consume(self.on_response, no_ack=True,
                                   queue=self.callback_queue)

    #callback method for when a response is received - note the check for correlation id
    def on_response(self, ch, method, props, body):
        if self.corr_id == props.correlation_id:
            self.response = body

    #method to make the initial request
    def call(self, n):
        self.response = None
        #generate a new correlation id
        self.corr_id = str(uuid.uuid4())

        #publish the message to the rpc_queue - note the reply_to property is set to the callback queue from above
        self.channel.basic_publish(exchange='',
                                   routing_key=self.rabbitqueue,
                                   properties=pika.BasicProperties(
                                         reply_to = self.callback_queue,
                                         correlation_id = self.corr_id,
                                         ),
                                   body=n)
        while self.response is None:
            self.rabbitconn.process_data_events()
        return self.response

print(" [.] query start time %r" % str(datetime.datetime.now()))
#instantiate an rpc client
crm_rpc = CrmRpcClient()

postreqdata = json.loads(open(os.environ['req']).read())
query = postreqdata['query']

crm_rpc = CrmRpcClient()
print(" [.] query start time %r" % str(datetime.datetime.now()))
queryresponse = crm_rpc.call(query)
print(" [.] query end time %r" % str(datetime.datetime.now()))
response = open(os.environ['res'], 'w')
response.write(queryresponse.decode())
response.close()

Here's a screenshot showing me calling the Python function with Postman.

Note the difference in time between the two functions - 5.62 seconds for Python and .46 seconds for C#!

Security and scalability

If you decide to use this approach in production, I'd suggest you carefully consider both security and scalability. Obviously the overall solution will only be as secure as your RabbitMQ broker and communications between the broker and its clients, so you'll want to look at best practices for access control and securing the communications with TLS. Here are some links for further reading on those subjects:

• TLS - https://www.rabbitmq.com/ssl.html
• Access control - https://www.rabbitmq.com/access-control.html

As for scalability, the approach I've shown creates a separate response queue for each consumer, but it can have problems scaling, especially if you are using a RabbitMQ cluster. You may want to look at the "direct reply-to" approach instead. For an interesting real-world overview of using direct reply-to, take a look at this blog post..

Wrapping up

I hope you've enjoyed this series and that it has given you some ideas about how to implement service relays in your Dynamics 365 CE projects. As I worked through the examples, I certainly learned a few new things, especially when I created my Python service proxy in Azure Functions.

Here are links to all the previous posts in this series.

1. Part 1 - Series introduction
2. Part 2 - Solution prerequisites
3. Part 3 - Python code for the consumer and listener processes

What do you think about this approach? Is it something you think you'd use in production? Let us know in the comments!

In my last post in this series, I walked through the prerequisites for building a simple service relay for Dynamics 365 CE with RabbitMQ and Python. In today's post I will show the Python code to make the service relay work.

As I described in the first post in this series, this approach relies on a consumer process and a queue listener process that can both access a RabbitMQ message broker.

A consumer writes a request to a cloud-hosted RabbitMQ request queue (either directly or through a proxy service) and starts waiting for a response. On the other end, a Python script monitors the request queue for inbound requests. When it sees a new one, it executes the appropriate request through the Dynamics 365 Web API and writes the response back to a client-specific RabbitMQ response queue. The consumer then picks up the response from the queue.

This solution is based on the remote procedure call (RPC) approach shown here. The main difference is that I have added logic to the queue monitoring script to query the Dynamics 365 Web API based on the inbound request from the consumer.

Consumer sample

The consumer does the following:

1. Read the text of the request to write to the queue from a command-line argument.
2. Establish a connection to the RabbitMQ broker.
3. Create a new anonymous, exclusive callback queue.
4. Write a request message a queue called "rpc_queue." This message will include the callback queue as its "reply_to" property.
5. Monitor the callback queue for a response.

There's no validation in this sample, so if you run it without a command-line argument, it will just throw an error and exit.

import sys
import pika
import uuid
import datetime

class CrmRpcClient(object):
    def __init__(self):
        #RabbitMQ connection details
        self.rabbituser = 'crmuser'
        self.rabbitpass = 'crmpass'
        self.rabbithost = '127.0.0.1' 
        self.rabbitport = 5672
        self.rabbitqueue = 'rpc_queue'
        rabbitcredentials = pika.PlainCredentials(self.rabbituser, self.rabbitpass)
        rabbitparameters = pika.ConnectionParameters(host=self.rabbithost,
                                    port=self.rabbitport,
                                    virtual_host='/',
                                    credentials=rabbitcredentials)

                self.rabbitconn = pika.BlockingConnection(rabbitparameters)

        self.channel = self.rabbitconn.channel()

        #create an anonymous exclusive callback queue
        result = self.channel.queue_declare(exclusive=True)
        self.callback_queue = result.method.queue

        self.channel.basic_consume(self.on_response, no_ack=True,
                                   queue=self.callback_queue)

    #callback method for when a response is received - note the check for correlation id
    def on_response(self, ch, method, props, body):
        if self.corr_id == props.correlation_id:
            self.response = body

    #method to make the initial request
    def call(self, n):
        self.response = None
        #generate a new correlation id
        self.corr_id = str(uuid.uuid4())

        #publish the message to the rpc_queue - note the reply_to property is set to the callback queue from above
        self.channel.basic_publish(exchange='',
                                   routing_key=self.rabbitqueue,
                                   properties=pika.BasicProperties(
                                         reply_to = self.callback_queue,
                                         correlation_id = self.corr_id,
                                         ),
                                   body=n)
        while self.response is None:
            self.rabbitconn.process_data_events()
        return self.response

#instantiate an rpc client
crm_rpc = CrmRpcClient()

#read the request from the command line
request = sys.argv[1]

#make the request and get the response
print(" [x] Requesting crm data("+request+")")
print(" [.] Start time %s" % str(datetime.datetime.now()))
response = crm_rpc.call(request)

#convert the response message body from the queue to a string 
decoderesponse = response.decode()

#print the output
print(" [.] Received response: %s" % decoderesponse)
print(" [.] End time %s" % str(datetime.datetime.now()))

Queue listener sample

The queue listener does the following:

1. Establish a connection to the RabbitMQ broker
2. Monitor "rpc_queue" queue.
3. When a new message from the "rpc_queue" queue is delivered, decode the message body as a string, and determine what Web API query to execute. Note: This sample can return a list of contacts or accounts from Dynamics 365 CE based on the request the consumer sends ("getcontacts" or "getaccounts"). If any other request is received, the listener will return an error message to the consumer callback queue.
4. Execute the appropriate query against the Dynamics 365 Web API and write the response to the callback queue the client established originally.

import pika
import requests
from requests_ntlm import HttpNtlmAuth
import json

#NTLM credentials to access on-prem Dynamics 365 Web API
username = 'DOMAIN\\USERNAME'
userpassword = 'PASSWORD'

#full path to Web API
crmwebapi = 'http://33.0.0.16/lucastest02/api/data/v8.1'

#RabbitMQ connection details
rabbituser = 'crmuser'
rabbitpass = 'crmpass'
rabbithost = '127.0.0.1' 
rabbitport = 5672

#method to execute a Web API query based on the client request
def processquery(query):
    #set the Web API request headers
    crmrequestheaders = {
        'OData-MaxVersion': '4.0',
        'OData-Version': '4.0',
        'Accept': 'application/json',
        'Content-Type': 'application/json; charset=utf-8',
        'Prefer': 'odata.maxpagesize=500',
        'Prefer': 'odata.include-annotations=OData.Community.Display.V1.FormattedValue'
    }

    #determine which Web API query to execute
    if query == 'getcontacts':
        crmwebapiquery = '/contacts?$select=fullname,contactid'
    elif query == 'getaccounts':
        crmwebapiquery = '/accounts?$select=name,accountid'
    else:
        #only handle 'getcontacts' or 'getaccounts' requests
        return 'Operation not supported'

    #execute the query
    crmres = requests.get(crmwebapi+crmwebapiquery, headers=crmrequestheaders,auth=HttpNtlmAuth(username,userpassword))
    
    #get the results json
    crmjson = crmres.json()

    #return the json
    return crmjson

#method to handle new inbound requests
def on_request(ch, method, props, body):
    #convert the message body from the queue to a string
    decodebody = body.decode('utf-8')

    #print the request
    print(" [.] Received request: '%s'" % decodebody)

    #process the request query
    response = processquery(decodebody)

    #publish the response back to 'reply-to' queue from the request message and set the correlation id
    ch.basic_publish(exchange='',
                     routing_key=props.reply_to,
                     properties=pika.BasicProperties(correlation_id = \
                                                         props.correlation_id),
                     body=str(response).encode(encoding="utf-8", errors="strict"))
    ch.basic_ack(delivery_tag = method.delivery_tag)

print(" [x] Awaiting RPC requests")

#connect to RabbitMQ broker
rabbitcredentials = pika.PlainCredentials(rabbituser, rabbitpass)
rabbitparameters = pika.ConnectionParameters(host=rabbithost,
                               port=rabbitport,
                               virtual_host='/',
                               credentials=rabbitcredentials)
rabbitconn = pika.BlockingConnection(rabbitparameters)
channel = rabbitconn.channel()

#declare the 'rpc_queue' queue
channel.queue_declare(queue='rpc_queue')

#set qos settings for the channel
channel.basic_qos(prefetch_count=1)

#assign the 'on_request' method as a callback for when new messages delivered from the 'rpc_queue' queue
channel.basic_consume(on_request, queue='rpc_queue')

#start listening for requests
channel.start_consuming()

Trying it out

As I mentioned in my last post, I initially wrote my code to use a RabbitMQ broker running on my local PC, so that's why the connections in the samples show 127.0.0.1 as the host. For a demo, I've spun up a copy of RabbitMQ in a Docker container in the cloud and updated my connection parameters accordingly, but I am still running my queue listener and consumer processes on my local PC.

When the listener first starts, it displays a simple status message.

Then I execute a "getcontacts" request from the consumer in a separate window.

From the timestamps before and after the request, you can see the round-trip time is less than .2 seconds, which includes two round trips between my local PC and the cloud-based RabbitMQ broker plus the actual query processing time in my local Dynamics 365 CE org.

Then I execute a "getaccounts" request.

This request was also fulfilled in less than .2 seconds.

Finally I execute an invalid request to show what the error response looks like.

You'll note the total time from request to response is only about .05 seconds less than the total time for the valid queries. That indicates most of the time used in these samples is being spent on the round trips between my local PC and the RabbitMQ broker, which is not surprising.

Meanwhile, the queue listener wrote a simple status update for every request it received. If I were using this in production, I would use more sophisticated logging.

Wrapping up

That's it for now. In my next (and final) post in this series, I will show how you can use Azure Functions to make a consumer service proxy so consuming applications don't have to access to your RabbitMQ broker directly, and I will also discuss some general thoughts on security and scalability for the service .

In my last post in this series, I outlined an approach for building a simple service relay with RabbitMQ and Python to easily expose an on-premises Dynamics 365 Customer Engagement organization to external consumers. In this post I will walk through the prerequisites for building this out. I'm assuming you have access to a Dynamics 365 CE organization, so I'm going to skip the setup for that and focus on just RabbitMQ and Python today.

Setting up RabbitMQ

Back in 2015 When I first blogged about RabbitMQ and Dynamics 365, I wrote a detailed post showing how to install and configure RabbitMQ. Since then I have discovered the joys of Docker, which makes the process significantly easier. If you have access to Docker, I highly recommend using it. Once you have Docker running, you can use one of the official RabbitMQ images. For this project, I initially used the rabbitmq:3-management image in Docker for Windows running on my local PC. After I got the basic functionality working, I then moved to an instance of Docker running in the cloud on a Digital Ocean VPS.

If don't want to use Docker, you can use a full RabbitMQ install like I showed previously. The main thing to remember is that no matter how you set up your RabbitMQ server, if it is not accessible from the public internet, you will not be able to use it as a service relay between an on-premises Dynamics 365 org and external consumers.

Setting up Python

I'm assuming if you've gotten this far, you have a functional Python development environment (if not, give Visual Studio Code a try), and the code I have written works in Python versions 2.7 or 3.x. In order to connect to both RabbitMQ and Dynamics 365, you will need a few additional packages. To connect to RabbitMQ, Pika is the RabbitMQ team's recommended Python client, and you can get it using pip.

To communicate with Dynamics 365, you'll need to use the Web API, but authentication will be handled differently depending on whether you connect to an on-premises org or an online / IFD org. For online or IFD orgs, you can either use ADAL or this alternate approach I described back in 2016. If you have an on-premises org, you can authenticate using the requests_ntlm package like I showed here. As with the Pika client, all the packages you need to connect to Dynamics 365 are also available via pip.

Wrapping up

That's it for today. In my next post in this series I will show the Python code you need to make this service relay work.

Integrating with external systems is a common requirement in Dynamics 365 Customer Engagement projects, but when the project involves an on-premises instance of Dynamics 365, routing requests from external systems through your firewall can present an additional challenge. Over the course of the next few posts, I will show you can easily build a simple service relay with RabbitMQ and Python to handle inbound requests from external data interface consumers.

Here's how my approach works. A consumer writes a request to a cloud-hosted RabbitMQ request queue (either directly or through a proxy service) and starts waiting for a response. On the other end, a Python script monitors the request queue for inbound requests. When it sees a new one, it executes the appropriate request through the Dynamics 365 Web API and writes the response back to a client-specific RabbitMQ response queue. The consumer then picks up the response from the queue. This way the consumer doesn't need to know anything other than how to write the initial request, and no extra inbound firewall ports need to be opened.

This diagram shows an overview of the process.

Although my original goal was to accelerate the deployment of data interfaces for on-premises Dynamics 365 CE instances, a simple service relay like this could also be useful for IFD or Dynamics 365 online deployments if you don't want to allow direct access to your organization. Because the queue monitoring process is single-threaded, it's an easy way to throttle requests, but you can run multiple instances of the queue monitor script if you want to increase the number of concurrent requests the relay can process.

Why use this approach?

There are lots message brokers and service bus offerings (Azure Service Bus, IBM MQ, Amazon SQS, etc.) you could use to build a service relay. In fact there's even an Azure offering called Azure Relay that aims to solve exactly the same problem that my approach does, but not just for Dynamics 365, so "why use this?" is a great question.

First, I think RabbitMQ is just a great tool, and I previously wrote a five-part series about using RabbitMQ with Dynamics 365 (back when it was still called Dynamics CRM). Second, using RabbitMQ instead of a cloud-specific service bus offering gives you maximum flexibility in where you host your request and response queues and how you chose to scale. For example, my RabbitMQ broker runs in a Docker container on a Digital Ocean VPS. If I ever decide to move off of Digital Ocean, I can easily switch to any IaaS or VPS provider. I can also configure a RabbitMQ cluster to achieve significantly faster throughput.

As for why I'm using Python instead of C#, which is probably more familiar to most Dynamics 365 developers, Python also makes this approach more flexible. Using Python means I'm not tied to the Dynamics 365 SDK client libraries or a Windows host for running my queue monitoring process, and I can easily package my monitoring process in a Docker image. (Although I highly recommend Python, there are RabbitMQ clients for .Net, and you can also find RabbitMQ tutorials for other languages including Java, Ruby and JavaScript here.)

Wrapping up

That's it for now. In my next post in this series I will walk through the prerequisites for building the simple service relay.

How have you handled inbound data interfaces for on-premises Dynamics 365 CE organizations? Let us know in the comments!

I've previously showed how to access online and IFD instances of Dynamics 365 Customer Engagement from Python code. Because that sample code authenticated to the Web API using OAuth, it won't work with on-premises instances. Recently I've been doing some work with Python and an on-premises Dynamics 365 organization, so I thought I'd share a sample that shows how to authenticate to the Web API using NTLM.

import requests
from requests_ntlm import HttpNtlmAuth
import json

username = 'companyx\\administrator'
userpassword = 'PASSWORD GOES HERE'

#set these values to query your crm data
crmwebapi = 'http://33.0.0.16/lucastest02/api/data/v8.1'
crmwebapiquery = '/contacts?$select=fullname,contactid'

crmrequestheaders = {
    'OData-MaxVersion': '4.0',
    'OData-Version': '4.0',
    'Accept': 'application/json',
    'Content-Type': 'application/json; charset=utf-8',
    'Prefer': 'odata.maxpagesize=500',
    'Prefer': 'odata.include-annotations=OData.Community.Display.V1.FormattedValue'
}

print('making crm request . . .')
crmres = requests.get(crmwebapi+crmwebapiquery, headers=crmrequestheaders,auth=HttpNtlmAuth(username,userpassword))
print('crm response received . . .')
try:
    print('parsing crm response . . .')
    crmresults = crmres.json()
    for x in crmresults['value']:
        print (x['fullname'] + ' - ' + x['contactid'])
except KeyError:
    print('Could not parse CRM results')

As you can see, this code doesn't retrieve an OAuth token before calling the Dynamics 365 Web API, but rather it uses the requests-ntlm package to authenticate directly to the Web API using a username and password. Other than that small change, everything else works the same as in my previous examples.

I've released an updated version of my popular Dynamics 365 Configuration Data Mover utility that includes the ability to create many-to-many associations in the target system. This upgrade is fully compatible with existing job files.

To create a many-to-many job step in the GUI, select the new "many to many" step type and input a FetchXML query for the relationship entity (relationship entity name on the many-to-many relationship form) that includes the GUID fields for each entity. The relationship entities cannot be queried in the advanced find builder, so you must write the FetchXML manually or use a separate query builder.

One thing to keep in mind is that a many-to-many job step will create many-to-many record associations in the target system, but it will not delete any existing N:N associations that have been removed in the source.

Getting the Dynamics 365 Configuration Data Mover

The source code is available in my GitHub repository here.

A compiled version can be downloaded here.