Skip to content

Instantly share code, notes, and snippets.

@thehappycheese
Last active November 9, 2023 09:08
Show Gist options
  • Save thehappycheese/25e798a620a488d268a8a0048992f0bc to your computer and use it in GitHub Desktop.
Save thehappycheese/25e798a620a488d268a8a0048992f0bc to your computer and use it in GitHub Desktop.
from pathlib import Path
import keyring
import pandas as pd
import os
import sys
# You need access to the compiled .NET AdomdClient .dll
# You can get it using nuget
# ```powershell
# nuget.exe install Microsoft.AnalysisServices.AdomdClient.retail.amd64
# nuget.exe install Microsoft.AnalysisServices.Tabular
# pip install pythonnet
# ```
# Alternatively you can just use the .net binaries distributed with DaxStudio
# (Which you should already be using!) by adding the bin folder to your path;
sys.path.append(os.path.expandvars(r"{PATH TO DAX STUDIO}\DaxStudio_3_0_10_portable\bin"))
# import from the Common Language Runtime;
import clr
clr.AddReference("Microsoft.AnalysisServices.AdomdClient")
from Microsoft.AnalysisServices.AdomdClient import AdomdConnection, AdomdCommand
# Create a client to convieniently hold the connection object
class SSASClient:
connection:AdomdConnection
def __init__(self, connection_string:str) -> None:
self.connection = AdomdConnection(connection_string)
self.connection.Open()
print("Connected to the cube successfully!")
def execute_mdx_query(self, mdx_command: str) -> pd.DataFrame:
command = AdomdCommand(mdx_command, self.connection)
reader = command.ExecuteReader()
result = []
while reader.Read():
sub_result = []
result.append(sub_result)
for i in range(reader.FieldCount):
try:
sub_result.append(reader[i])
except:
break
field_names = [reader.GetName(i) for i in range(reader.FieldCount)]
reader.Close()
df = pd.DataFrame(
columns=field_names,
data=result
)
return df
# example usage (fetching connection string from windows credential manager or similar using keyring)
client = SSASClient(keyring.get_password("cube_warehouse","user"))
cubes = client.execute_mdx_query(
"SELECT * FROM $SYSTEM.MDSCHEMA_CUBES WHERE CUBE_SOURCE = 1"
)
print(cubes)
# You can get nice stuff with this approach by hooking into PowerBI
from System import String, Int32
from Microsoft.AnalysisServices.AdomdClient import AdomdConnection, AdomdCommand
from Microsoft.AnalysisServices.Tabular import DataType
# utility to map data type names later
def get_data_type_name(enum_value:int) -> str:
return DataType.GetName(DataType, Int32(enum_value))
# If you have DAX Studio you can add its path to the .env file (the .../bin directory)
clr.AddReference("Microsoft.AnalysisServices.Tabular")
clr.AddReference("DaxStudio.UI")
from DaxStudio.UI.Utils import PowerBIHelper
items = PowerBIHelper().GetLocalInstances(True)
powerbi_instance = items[0]
# get the port number for the first instance of powerbi found.
# will error if none found
MSMDSRVEXE_PORT = powerbi_instance.Port
print(f"Found first PowerBI instance with window title:\n {powerbi_instance.Name}")
print(f"The model is being served by msmgdsrv.exe on port:\n {MSMDSRVEXE_PORT}")
client = SSASClient(f"Provider=MSOLAP;Data Source=127.0.0.1:{MSMDSRVEXE_PORT};")
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment