Obtener listado de Dossiers y Datasets con Python

En el siguiente ejemplo vamos a comprobar cómo obtener listado de Dossiers y Datasets con Python en pocos pasos.

Al final del artículo puedes descargarte el script para Jupyter notebook o si lo prefieres, puedes descargar el código python para ejecutarlo directamente en tu consola de Workstation y programar su ejecución (el código entre Jupyter notebook y workstation tienen ligeras diferencias ya que la consola de Workstation no es 100% compatible, pero con el mismo resultado final)

Especificaciones del entorno

• Microsoft Windows 10
• MicroStrategy 2021 update 6 (11.3.6)
• Mstrio 11.3.6.102
• Python v3.9
• Jupyter Notebook

Demostración en video

Explicación del código

Importar librerías

Lo primero que realizamos es la importación de todas las librerías necesarias que van a intervenir en nuestro script.

import csv
import pandas as pd
from datetime import datetime
from mstrio.types import ObjectSubTypes, ObjectTypes
from mstrio.project_objects import Dossier
from mstrio.project_objects.dossier import list_dossiers_across_projects
from mstrio.object_management import Object

Conexión

El siguiente paso sera realizar la conexión mediante MicroStrategy Library y definir los datos del servidor y del proyecto. En esta parte debes realizar cambios para adaptarlo a tu servidor.

import getpass
from mstrio.connection import Connection

mstr_usuario = "Administrator"
mstr_pass = getpass.getpass(prompt='Password ')
mstr_base_url = "https://desktop-5dqnoai:8443"
mstr_url_api  = mstr_base_url+"/MicroStrategyLibrary/api"
projecto_id    = "B7CA92F04B9FAE8D941C3E9B7E0CD754"

conn = Connection(mstr_url_api, mstr_usuario, mstr_pass, project_id=projecto_id, ssl_verify=False)

• Modifica tu usuario de MicroStrategy en la variable mstr_usuario
• En la línea 4, la variable mstr_pass preguntara en cada ejecución por la contraseña del usuario (no compatible en la consola de Workstation, pero sí en jupyter notebook). Lo puedes borrar y escribir tu contraseña directamente.
• En la línea 5, modifica la dirección del servidor IS.
• En la línea 6, la dirección de Library.
• Si tu objetivo es obtener un listado de todos los proyectos elimina la fila 7 y en la ultima elimina project_id=projecto_id, de lo contrario cambia el ID del proyecto.
• El parámetro de conexión ssl_verify es opcional, en mi caso lo incluyo al ser un entorno de pruebas.

Declaración de funciones

Se declaran funciones para después reutilizarlas como son la búsqueda, la obtención de datasets y de dossiers.

def search_ancestors(ancestors):
    path=''
    if ancestors: 
        for i in range(len(ancestors)):
            path= path+' > '+(ancestors[i]['name'])
    else:
        path="___NO PATH"
    return path[3:]

def get_datasets(connection, project_ID, dossier_ID):
    res = connection.get(url=f"{connection.base_url}/api/v2/dossiers/{dossier_ID}/definition",
                                headers={'X-MSTR-ProjectID': project_ID})
    return res

def get_object(connection, project_ID, object_ID):
    res = connection.get(url=f"{connection.base_url}/api/objects/{object_ID}?type=3",
                                headers={'X-MSTR-ProjectID': project_ID})
    return res

Buscar dossiers

El siguiente paso es obtener el numero de dossiers que ha encontrado.

ldap=list_dossiers_across_projects(connection=conn)
print(f"Nº de dossiers: {len(ldap)}")

Información de dossiers

Ahora de cada dossier traemos cierta información como su ID, nombre, ruta, etc junto a los datasets que se utilizan en cada dossier.

print("\nEjecutando... por favor, espere.\n")

count_dos=0  # this counter will give you hints how long the whole process will take
dt1=datetime.now()
dossiers_list,datasets_list=[],[]

for a in ldap[:]:        
    datasets_json=get_datasets(conn, a._project_id, a.id).json()
    datasets_local=[["missing","missing"]]
    
    if djd:=datasets_json.get('datasets'): # walrus operator; works for Python 3.8 or higher
        datasets_local=[]
        for d in djd:
            datasets_local.append([d['name'], d['id']]) #this is just for this dossier
            datasets_list.append((d['name'], d['id'], a._project_id)) #this gathers all datasets
    for dt in datasets_local:
        dossiers_list.append([a._project_id, a.id, a.name, search_ancestors(a._ancestors),
                              a._date_created.strftime("%Y-%m-%d %H:%M:%S"),
                              a._date_modified.strftime("%Y-%m-%d %H:%M:%S"),
                              dt[0], dt[1]])        
    
    count_dos+=1
    if (count_dos % 50) == 0 or count_dos == (len(ldap)):
        eta = (len(ldap[:])/count_dos)*(datetime.now()-dt1)
        print(count_dos, "dossiers procesados: ", eta)
    #print(a.__dict__)

print("\nFin.\n")
print(f"Nº datasets: {len(dossiers_list)} (algunos estan vacios, otros se usan varias veces)")
datasets_list=list(set(datasets_list))
print(f"Datasets unicos: {len(datasets_list)}")

Exportar resultados dossiers

En este paso exportamos los resultados de los dossiers a un CSV

headers_dossiers=["ID Proyecto", "ID Dossier","Nombre Dossier", "Ruta Dossier", "Fecha creación", "Fecha modificación", 
                  "Nombre Dataset", "Id Dataset"]
df = pd.DataFrame(dossiers_list, columns = headers_dossiers)
df.to_csv(f"lista_de_dossiers.csv", index=False, encoding='utf-8', sep=",", 
          quoting=csv.QUOTE_ALL, escapechar="\\", header=True)

print("\nFin de la exportación.\n")

Información de dataset

En este paso realizamos casi lo mismo pero para obtener información de cada dataset.

print("\nEjecutando... por favor, espere.\n")

count_dos=0  # this counter will give you hints how long the whole process will take
dt1=datetime.now()
datasets_list_new=[]

for b in datasets_list[:]:        
    dj=get_object(conn, b[2], b[1]).json()
    datasets_list_new.append([b[2], dj["id"], dj["name"], dj["type"], dj["subtype"], 
                              dj["dateCreated"], dj["dateModified"], search_ancestors(dj["ancestors"])])
    
    count_dos+=1
    if (count_dos % 50) == 0 or count_dos == (len(datasets_list)):
        eta = (len(datasets_list[:])/count_dos)*(datetime.now()-dt1)
        print(count_dos, "datasets procesados: ", eta)
        
print("\nFin.\n")

Exportar resultados dataset

En este paso exportamos los resultados de los dataset a un CSV

headers_datasets=["ID Proyecto","Id Dataset", "Nombre Dataset", "Tipo Dataset","Subtipo Dataset",
                  "Fecha creación","Fecha modificación","Ruta Dataset"]
df = pd.DataFrame(datasets_list_new, columns = headers_datasets)
df.to_csv(f"lista_de_datasets.csv", index=False, encoding='utf-8', sep=",", 
          quoting=csv.QUOTE_ALL, escapechar="\\", header=True)

print("\nFin de la exportación.\n")

Subtipos de dataset

Ten encuenta esta lista de subtipos de dataset para entender mejor los resultados:

Descarga de ejemplos