Beginner+ Python Script

[ChristiaanEmpowerBI]

This solution requires a Dossier in the Tutorial Project with a filter and a visualization with Customer on it. It prompts the user for a Customer ID.

I know the code might be made better, and if anyone has any suggestions, please feel free to comment on criticize.

#pip install mstrio.py
#https://github.com/MicroStrategy/mstrio-py
import mstrio
from mstrio.connection import get_connection, Connection
from mstrio.utils.error_handlers import ErrorHandler
import re
import requests
import json

PROJECT_NAME = 'Tutorial Project'
BASE_URL = 'http://my_pc:8080/MicroStrategyLibrary/api'
MSTR_USERNAME = 'Administrator'
MSTR_PASSWORD = ''
project_id = 'B19DEDCC11D4E0EFC000EB9495D0F44F'
dossier_id = 'C3D32F6C48EC04BACC94D281C4ABC936'
cube_id = '8E0475AF4BD97ECC9C618189A9493A23'
customer_attribute_id = '8D679D3C11D3E4981000E787EC6DE8A4'

conn = Connection(
BASE_URL,
MSTR_USERNAME,
MSTR_PASSWORD,
project_id=project_id,
login_mode=1
)

search_customer = input('Customer ID:')

seperator = '-' * 100
seperator_api = seperator + ' API Call'

def login(connection):
return connection.post(
skip_expiration_check=True,
endpoint='/api/auth/login',
data={
'username': connection.username,
'password': connection._Connection__password,
'loginMode': connection.login_mode
},
)

def get_JSESSIONID(input_string):
pattern = r'JSESSIONID=([^;]+)'
match = re.search(pattern, input_string)
if match:
JSESSIONID = match.group(1)
return JSESSIONID
else:
print('No JSESSIONID found.')

def GET_api_dossiers_dossierid_definition(auth_token, project_id, dossier_id, connection):
api_request = f'/api/dossiers/{dossier_id}/definition'
print(seperator_api)
print('GET:', api_request)
print(seperator)
request_headers={'accept' : 'application/json', 'X-MSTR-AuthToken': auth_token, 'X-MSTR-ProjectID': project_id, }
return connection.get(
endpoint=api_request,
headers=request_headers,
)

def POST_api_dossiers_dossierid_instances(auth_token, project_id, dossier_id, element_id, connection):
filter_key = 'W2FB6A76DE9734A3D8D9004FC482AC3B3'
'''
{
"filters": [
{
"key": "W2FB6A76DE9734A3D8D9004FC482AC3B3",
"selections": [
{
"name": "7796"
}
]
}
],
"persistViewState": true
}
'''
api_request = f'/api/dossiers/{dossier_id}/instances'
print(seperator_api)
print('POST:', api_request)
print(seperator)
request_headers={'accept' : 'application/json', 'X-MSTR-AuthToken': auth_token, 'X-MSTR-ProjectID': project_id, }
request_data = {
'filters': [
{
'key': filter_key,
'selections': [
{
'id': f'{element_id}'
}
]
}
],
'persistViewState': 'true'
}
print(request_data)
response = connection.post(
endpoint=api_request,
headers=request_headers,
json=request_data
)

# Check the Response Content-Type
if 'application/json' in response.headers.get('Content-Type', ''):
    try:
        response_body = response.json()
    except Exception as e:
        print(f"Error parsing JSON: {e}")
        response_body = response.text
else:
    response_body = response.text

# Check the HTTP response status code
print("HTTP Response Status Code:", response.status_code)

return response.headers, response_body

def GET_api_dossiers_dossierid_instances_dossierinstanceid_elements(auth_token, project_id, dossier_id, dossier_instance_id, connection):
target_object_id = '8D679D3C11D3E4981000E787EC6DE8A4' # Customer Attribute ID
filter_key_customer = 'W2FB6A76DE9734A3D8D9004FC482AC3B3'
filter_key_status = 'WCD4678F9C7F5476898BDB6213EC95A4E'
print(seperator_api)
#api_request = f'/api/dossiers/{dossier_id}/instances/{dossier_instance_id}/elements?targetObjectId=8D679D3C11D3E4981000E787EC6DE8A4&targetObjectType=attribute'
api_request = f'/api/dossiers/{dossier_id}/instances/{dossier_instance_id}/elements?targetObjectId={target_object_id}&targetObjectType=attribute&searchPattern={search_customer}&filterKey={filter_key_customer}'
print('GET:', api_request)
print(seperator)
request_headers={'accept' : 'application/json', 'X-MSTR-AuthToken': auth_token, 'X-MSTR-ProjectID': project_id, }
return connection.get(
endpoint=api_request,
headers=request_headers,
)

def GET_api_cubes_cubeid_attributes_attributeid_elements(auth_token, project_id, cube_id, attribute_id, connection):
print(seperator_api)
api_request = f'/api/cubes/{cube_id}/attributes/{attribute_id}/elements?offset=0&limit=-1'
print('GET:', api_request)
print(seperator)
request_headers={'accept' : 'application/json', 'X-MSTR-AuthToken': auth_token, 'X-MSTR-ProjectID': project_id, }
return connection.get(
endpoint=api_request,
headers=request_headers,
)

def GET_api_projects(auth_token, connection):
api_request = f'/api/projects'
request_headers={'accept' : 'application/json', 'X-MSTR-AuthToken': auth_token,}
return connection.get(
endpoint=api_request,
headers=request_headers,
)
def GET_api_dossiers_dossierid_instances_instanceid_chapters_chapterkey_visualizations_vizualizationKey(auth_token, project_id, dossier_id, dossier_instance_id, chapterKey, vizualizationKey, connection):
api_request = f'/api/dossiers/{dossier_id}/instances/{dossier_instance_id}/chapters/{chapterKey}/visualizations/{vizualizationKey}'
print(seperator_api)
print('GET:', api_request)
print(seperator)
request_headers={'accept' : 'application/json', 'X-MSTR-AuthToken': auth_token,}
return connection.get(
endpoint=api_request,
headers=request_headers,
)

def print_project_ids(dictionary_list):
for item in dictionary_list:
if 'id' in item:
print('Project Name:', item['name'])
print('Project ID:', item['id'])

def convert_bytes_to_JSON(bytes_input):
json_string = bytes_input.decode('utf-8')
data_dictionary = json.loads(json_string)
return data_dictionary

def print_POST_api_dossiers_dossierid_instances(json_data):
print("instance------------------------------------")
print("mid:", json_data.get("mid", ""))
print("status:", json_data.get("status", ""))
print("jobId:", json_data.get("jobId", ""))

def extract_chapters_pages_visualizations_filters(json_data):
chapters_list = []
pages_list = []
visualizations_list = []
filters_list = []

# Extract chapters
chapters = json_data.get("chapters", [])
for chapter in chapters:
    chapter_info = {"key": chapter["key"], "name": chapter["name"]}
    chapters_list.append(chapter_info)

    # Extract pages
    pages = chapter.get("pages", [])
    for page in pages:
        page_info = {"key": page["key"], "name": page["name"]}
        pages_list.append(page_info)

        # Extract visualizations
        visualizations = page.get("visualizations", [])
        for visualization in visualizations:
            visualization_info = {"key": visualization["key"], "name": visualization["name"]}
            visualizations_list.append(visualization_info)

    # Extract filters
    filters = chapter.get("filters", [])
    for filter_item in filters:
        filter_info = {"key": filter_item["key"], "name": filter_item["name"]}
        filters_list.append(filter_info)

return chapters_list, pages_list, visualizations_list, filters_list   

def print_GET_api_dossiers_dossierid_definition_response(json_data):
# Access and print the "chapters"
chapters = json_data.get("chapters", [])
for chapter in chapters:
print("Chapter-------------------------------------")
print("key:", chapter.get("key", ""))
print("name:", chapter.get("name", ""))

    # Access and print the "pages" within the chapter
    pages = chapter.get("pages", [])
    for page in pages:
        print("Page----------------------------------------")
        print("key:", page.get("key", ""))
        print("name:", page.get("name", ""))
    
    # Access and print the "filters" within the chapter
    filters = chapter.get("filters", [])
    for filter_item in filters:
        print("Filter--------------------------------------")
        print("key:", filter_item.get("key", ""))
        print("name:", filter_item.get("name", ""))
        print("summary:", filter_item.get("summary", ""))
        source = filter_item.get("source", {})
        print("source:")
        print("id:", source.get("id", ""))
        print("type:", source.get("type", ""))
print('----------------------------------------------------------------------------------------')

def find_elementid(json_data, customer_id):
for item in json_data:
item_id = item["id"]
# Split the "id" using ":" as the delimiter
parts = item_id.split(":")
#print('len(parts):', len(parts))
#print('parts[1]', parts[1])
if len(parts) == 2 and parts[1] == customer_id:
element_id = f'h{customer_id};{parts[0]}'
return element_id
return None # Return None if not found

####################### Step 1, POST /api/auth/apiTokens ##########################
####################### connect to the Intelligence Server ##########################
HTTP_response = login(conn) #<class 'requests.models.Response'>
JSESSIONID = get_JSESSIONID(HTTP_response.headers['Set-Cookie'])
print('JSESSIONID:',JSESSIONID)
X_MSTR_AuthToken = HTTP_response.headers['X-MSTR-AuthToken']
print('X-MSTR-AuthToken:', X_MSTR_AuthToken)
print(seperator)

####################### Step 2, GET /api/cubes/{cubeId}/attributes/{attributeId}/elements #####################
####################### get the element id and name to seatch for in the filter #####################
HTTP_response = GET_api_cubes_cubeid_attributes_attributeid_elements(X_MSTR_AuthToken, project_id, cube_id, customer_attribute_id, conn) #<class 'requests.structures.CaseInsensitiveDict'>
response_JSON = convert_bytes_to_JSON(HTTP_response.content)
formatted_json = json.dumps(response_JSON, indent=2)
#print(formatted_json)
element_id = find_elementid(response_JSON, search_customer)
print('element_id:', element_id)

####################### Step 3, GET /api/dossiers/{dossierId}/definition #####################
####################### get the dossier definition #####################
####################### to get filter key, vizualization key #####################
HTTP_response = GET_api_dossiers_dossierid_definition(X_MSTR_AuthToken, project_id, dossier_id, conn) #<class 'requests.structures.CaseInsensitiveDict'>
response_JSON = convert_bytes_to_JSON(HTTP_response.content)
formatted_json = json.dumps(response_JSON, indent=2)
#print(formatted_json)
#print_GET_api_dossiers_dossierid_definition_response(response_JSON)
chapters, pages, visualizations, filters = extract_chapters_pages_visualizations_filters(response_JSON)
#print(chapters)
first_chapters_dictionary = chapters[0]
chapterKey = first_chapters_dictionary['key']
#print('chapterKey:', chapterKey)
#print(visualizations)
first_visualizations_dictionary = visualizations[0]
vizualizationKey = first_visualizations_dictionary['key']
#print('visualizationKey:', vizualizationKey)

####################### Step 4, POST /api/dossiers/{dossierId}/instances #####################
####################### create an instance of a dossier #####################
####################### filter key: W2FB6A76DE9734A3D8D9004FC482AC3B3 #####################
HTTP_response_header, HTTP_response_body = POST_api_dossiers_dossierid_instances(X_MSTR_AuthToken, project_id, dossier_id, element_id, conn) #<class 'requests.structures.CaseInsensitiveDict'>
#print('HTTP_response_header:', HTTP_response_header)
#print('HTTP_response_body:', HTTP_response_body)
#response_JSON = convert_bytes_to_JSON(HTTP_response.content)
dossier_instance_id = HTTP_response_body.get('mid', '')#print('Dossier Instance ID:', dossier_instance_id)
print('Dossier Instance ID:', dossier_instance_id)
#formatted_json = json.dumps(response_JSON, indent=2)
#print(formatted_json)

####################### Step 5, GET api/dossiers/{dossierId}/instances/{instanceId}/chapters/{chapterKey}/visualizations/{visualizationKey} ##########################
####################### get the data ##########################
HTTP_response = GET_api_dossiers_dossierid_instances_instanceid_chapters_chapterkey_visualizations_vizualizationKey(X_MSTR_AuthToken, project_id, dossier_id, dossier_instance_id, chapterKey, vizualizationKey, conn)
response_JSON = convert_bytes_to_JSON(HTTP_response.content)
formatted_json = json.dumps(response_JSON, indent=2)
#print(formatted_json)

Access the 'result' section

result_data = response_JSON["result"]

Access the 'data' section within 'result'

data = result_data["data"]

Access the 'root' section within 'data'

root = data["root"]

Access the 'children' list within 'root'

children = root["children"]

Loop through the 'children' list

for child in children:
# Access and print the 'formValues' section containing "ID" and "Last Name"
form_values = child["element"]["formValues"]
print("ID:", form_values.get("ID"))
print("Last Name:", form_values.get("Last Name"))
# Access and print the 'metrics' section
metrics = child["metrics"]
for metric_name, metric_data in metrics.items():
print(metric_name + " - rv:", metric_data["rv"])
# Add a separator for readability
print(seperator)

conn.close()