Skip to content

Instantly share code, notes, and snippets.

@mjsqu
Created January 28, 2024 20:31
Show Gist options
  • Save mjsqu/bf4a17a7b03d64ca21d0d76b634dc32f to your computer and use it in GitHub Desktop.
Save mjsqu/bf4a17a7b03d64ca21d0d76b634dc32f to your computer and use it in GitHub Desktop.
Dataverse MSAL pyodata - pyodata not compatible with odata v4 (odata >v2)
import msal
import pyodata
"""
The configuration file would look like this (sans those // comments):
{
"authority": "https://login.microsoftonline.com/Enter_the_Tenant_Name_Here",
"client_id": "your_client_id",
"scope": ["https://graph.microsoft.com/.default"],
// For more information about scopes for an app, refer:
// https://docs.microsoft.com/en-us/azure/active-directory/develop/v2-oauth2-client-creds-grant-flow#second-case-access-token-request-with-a-certificate"
"secret": "The secret generated by AAD during your confidential app registration",
// For information about generating client secret, refer:
// https://github.com/AzureAD/microsoft-authentication-library-for-python/wiki/Client-Credentials#registering-client-secrets-using-the-application-registration-portal
"endpoint": "https://graph.microsoft.com/v1.0/users"
}
You can then run this sample with a JSON configuration file:
python sample.py parameters.json
"""
import sys # For simplicity, we'll read config file from 1st CLI param sys.argv[1]
import json
import logging
import requests
# Optional logging
# logging.basicConfig(level=logging.DEBUG)
config = json.load(open(sys.argv[1]))
scopes = [f"{config['resource']}/.default"]
# Create a preferably long-lived app instance which maintains a token cache.
app = msal.ConfidentialClientApplication(
config["client_id"], authority=f"https://login.microsoftonline.com/{config['tenant_id']}",
client_credential=config["client_secret"],
# token_cache=... # Default cache is in memory only.
# You can learn how to use SerializableTokenCache from
# https://msal-python.rtfd.io/en/latest/#msal.SerializableTokenCache
)
# The pattern to acquire a token looks like this.
result = None
# Firstly, looks up a token from cache
# Since we are looking for token for the current app, NOT for an end user,
# notice we give account parameter as None.
result = app.acquire_token_silent(scopes, account=None)
if not result:
logging.info("No suitable token exists in cache. Let's get a new one from AAD.")
result = app.acquire_token_for_client(scopes=scopes)
if "access_token" in result:
token = result['access_token']
# Calling graph using the access token
dataverse_data = requests.get( # Use token to call downstream service
f"{config['resource']}/api/data/v9.2/$metadata",
headers={'Authorization': 'Bearer ' + token},
)
else:
print(result.get("error"))
print(result.get("error_description"))
print(result.get("correlation_id")) # You may need this when reporting a bug
service_url = f"{config['resource']}/api/data/v9.2/"
import pyodata
session = requests.Session()
session.headers.update({"Authorization": token})
namespaces = {
"edmx": "http://docs.oasis-open.org/odata/ns/edmx",
"edm": "http://docs.oasis-open.org/odata/ns/edm"
}
from pyodata.v2.model import PolicyFatal, PolicyWarning, PolicyIgnore, ParserError, Config
custom_config = Config(
xml_namespaces=namespaces,
default_error_policy=PolicyFatal(),
custom_error_policies={
ParserError.ANNOTATION: PolicyWarning(),
ParserError.ASSOCIATION: PolicyIgnore(),
ParserError.ENUM_TYPE: PolicyIgnore(),
},
retain_null=True)
service_client = pyodata.Client(service_url, session)
# pyodata is incompatible with OData V4 - https://github.com/SAP/python-pyodata/issues/140
# msal works though - this should be how we connect our tap
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment