|
import pyodbc |
|
import msal |
|
import requests |
|
import json |
|
import re |
|
import time |
|
|
|
# Azure AD details |
|
client_id = '0e1c58b1-3d9a-4618-8889-6c6505288d3c' |
|
client_secret = 'qlU8Q~dmhKFfdL1ph2YsLK9URbhIPn~qWmfr1ceL' |
|
tenant_id = '97ae7e35-2f87-418b-9432-6733950f3d5c' |
|
authority = f'https://login.microsoftonline.com/{tenant_id}' |
|
resource = 'https://ecellorsdev.crm8.dynamics.com' |
|
|
|
# SQL endpoint |
|
sql_server = 'ecellorsdev.crm8.dynamics.com' |
|
database = 'ecellorsdev' |
|
|
|
# Get token with error handling |
|
try: |
|
print(f"Attempting to authenticate with tenant: {tenant_id}") |
|
print(f"Authority URL: {authority}") |
|
|
|
app = msal.ConfidentialClientApplication(client_id, authority=authority, client_credential=client_secret) |
|
|
|
print("Acquiring token…") |
|
|
|
token_response = app.acquire_token_for_client(scopes=[f'{resource}/.default']) |
|
|
|
if 'error' in token_response: |
|
print(f"Token acquisition failed: {token_response['error']}") |
|
print(f"Error description: {token_response.get('error_description', 'No description available')}") |
|
else: |
|
access_token = token_response['access_token'] |
|
print("Token acquired successfully and your token is!"+access_token) |
|
print(f"Token length: {len(access_token)} characters") |
|
|
|
except ValueError as e: |
|
print(f"Configuration Error: {e}") |
|
|
|
except Exception as e: |
|
print(f"Unexpected error: {e}") |
|
|
|
#Get 5 contacts from Dataverse using Web API |
|
import requests |
|
import json |
|
|
|
try: |
|
#Full CRUD Operations – Create, Read, Update, Delete a contact in Dataverse |
|
print("Making Web API request to perform CRUD operations on contacts…") |
|
|
|
# Dataverse Web API endpoint for contacts |
|
web_api_url = f"{resource}/api/data/v9.2/contacts" |
|
|
|
# Base headers with authorization token |
|
headers = { |
|
'Authorization': f'Bearer {access_token}', |
|
'OData-MaxVersion': '4.0', |
|
'OData-Version': '4.0', |
|
'Accept': 'application/json', |
|
'Content-Type': 'application/json' |
|
} |
|
|
|
# Simple approach: create multiple contacts sequentially |
|
# generate 100 contacts with different last names |
|
contacts_to_create = [ |
|
{"firstname": "Ecellors", "lastname": f"Test{str(i).zfill(3)}"} |
|
for i in range(1, 101) |
|
] |
|
|
|
create_headers = headers.copy() |
|
create_headers['Prefer'] = 'return=representation' |
|
|
|
created_ids = [] |
|
print("Creating contacts sequentially…") |
|
for i, body in enumerate(contacts_to_create, start=1): |
|
try: |
|
resp = requests.post(web_api_url, headers=create_headers, json=body, timeout=15) |
|
except requests.exceptions.RequestException as e: |
|
print(f"Request error creating contact #{i}: {e}") |
|
continue |
|
|
|
if resp.status_code in (200, 201): |
|
try: |
|
j = resp.json() |
|
cid = j.get('contactid') |
|
except ValueError: |
|
cid = None |
|
|
|
if cid: |
|
created_ids.append(cid) |
|
print(f"Created contact #{i} with id: {cid}") |
|
|
|
else: |
|
print(f"Created contact #{i} but response body missing id. Response headers: {resp.headers}") |
|
|
|
elif resp.status_code == 204: |
|
# try to extract id from headers |
|
entity_url = resp.headers.get('OData-EntityId') or resp.headers.get('Location') |
|
|
|
if entity_url: |
|
m = re.search(r"([0-9a-fA-F\-]{36})", entity_url) |
|
if m: |
|
cid = m.group(1) |
|
created_ids.append(cid) |
|
print(f"Created contact #{i} (204) with id: {cid}") |
|
else: |
|
print(f"Created contact #{i} (204) but couldn't parse id from headers: {resp.headers}") |
|
else: |
|
print(f"Created contact #{i} (204) but no entity header present: {resp.headers}") |
|
else: |
|
print(f"Failed to create contact #{i}. Status code: {resp.status_code}, Response: {resp.text}") |
|
|
|
# small pause to reduce chance of throttling/rate limits |
|
time.sleep(0.2) |
|
|
|
if created_ids: |
|
print("Created contact ids:") |
|
for cid in created_ids: |
|
print(cid) |
|
|
|
except Exception as e: |
|
print(f"Unexpected error during Execute Multiple: {e}") |
|
print("Failed to extract Contact ID from headers.") |
|
|