Skip to content

Base SSAS Server#

Base Server Interface containing the methods used outside of instance lifetime management.

Source code in pbi_core/ssas/server/server.py 
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
class BaseServer:
    """Base Server Interface containing the methods used outside of instance lifetime management."""

    host: str
    "A hostname to the background SSAS instance"

    port: int
    "A Port to the background SSAS instance"

    default_db: str | None
    """The default DB to use when executing DAX"""

    def __init__(self, host: str, port: int, default_db: str | None = None) -> None:
        self.host = host
        self.port = port
        self.default_db = default_db

        self.check_ssas_sku()

    def conn_str(self, db_name: str | None = None) -> str:
        """Formats the connection string for connecting to the background SSAS instance."""
        if db_name:
            return f"Provider=MSOLAP;Data Source={self.host}:{self.port};Initial Catalog={db_name};"
        return f"Provider=MSOLAP;Data Source={self.host}:{self.port};"

    def conn(self, db_name: str | None = None) -> pbi_pyadomd.Connection:
        """Returns a pbi_pyadomd connection."""
        return pbi_pyadomd.connect(self.conn_str(db_name))

    def __repr__(self) -> str:
        return f"Server(host={self.host}:{self.port})"

    def query_dax(self, query: str, *, db_name: str | bool | None = True) -> list[dict[str, Any]]:
        """db_name: when bool and == True, uses the DB last loaded by this server instance.

        (almost always the db of the loaded PBI unless you're manually reassigning server instances)
        when None or False, no db is supplied
        when a string, just passed to the client
        """
        if db_name is True:
            db_name = self.default_db
        elif db_name is False:
            db_name = None
        with self.conn(db_name) as conn:
            reader = conn.execute_dax(query)
            return reader.fetch_many()

    def query_xml(self, query: str, db_name: str | None = None) -> BeautifulSoup:
        """Submits an XMLA query to the SSAS instance and returns the result as a BeautifulSoup object.

        The query should be a valid XMLA command.

        Args:
            query (str): The XMLA query to execute.
            db_name (str | None): The name of the database to execute the query against.

        Returns:
            BeautifulSoup: The result of the query parsed as a BeautifulSoup object.

        """
        with self.conn(db_name) as conn:
            return conn.execute_xml(query)

    def tabular_models(self) -> list[BaseTabularModel]:
        """Creates a list of the Tabular models existing in the SSAS server.

        Note:
            Generally tabular models in the local environment correspond 1-1 with a PBIX report open in the Desktop app

        """
        # Query based on https://learn.microsoft.com/en-us/previous-versions/sql/sql-server-2012/ms126314(v=sql.110)
        dbs = self.query_dax(COMMAND_TEMPLATES["list_dbs.xml"].render())
        return [BaseTabularModel(row["CATALOG_NAME"], self) for row in dbs]

    @backoff.on_exception(backoff.expo, ValueError, max_time=10)
    def check_ssas_sku(self) -> None:
        """Checks if the SSAS instance is running the correct SKU version.

        Tests this assumption by running a query that should fail if the SKU under 1400 (image save command).
        Since we could also fail due to the server not being instantiated, we use a backoff decorator
        to retry the command a few times before giving up.

        Raises:
            TypeError: If the SSAS instance is running an incorrect SKU version.
            ValueError: If the SSAS instance is not running or the command fails for another reason.
                This is used to trigger the backoff retry.

        """
        try:
            self.query_xml(
                COMMAND_TEMPLATES["image_save.xml"].render(
                    target_path="---",
                    db_name="---",
                ),  # specifically choosing non-existant values to verify we get at least one error
            )
        except pbi_pyadomd.AdomdErrorResponseException as e:
            error_message = str(e.Message)
            if error_message == SKU_ERROR:
                return
            msg = f"Incorrect SKUVersion. We got the error: {error_message}"
            raise TypeError(msg) from None
        msg = "Got a 'file not loaded' type error. Waiting"
        raise ValueError(msg)

    @staticmethod
    def sanitize_xml(xml_text: str) -> str:
        """Method to XML-encode characters like "&" so that the Adomd connection doesn't mangle the XMLA commands."""
        return xml_text.replace("&", "&")

    @staticmethod
    def remove_invalid_db_name_chars(orig_db_name: str) -> str:
        """Utility function to convert a PBIX report name to an equivalent name for the DB in the SSAS instance.

        Note:
            Raises a warning if the db_name is changed to inform user that the db_name does not match their input

        """
        db_name = orig_db_name.replace("&", " ")[:100]
        db_name = db_name.strip()  # needed to find the correct name, since SSAS does stripping too
        if orig_db_name != db_name:
            logger.warning("db_name changed", original_name=orig_db_name, new_name=db_name)
        return db_name

default_db instance-attribute #

default_db: str | None = default_db

The default DB to use when executing DAX

host instance-attribute #

host: str = host

A hostname to the background SSAS instance

port instance-attribute #

port: int = port

A Port to the background SSAS instance

check_ssas_sku #

check_ssas_sku() -> None

Checks if the SSAS instance is running the correct SKU version.

Tests this assumption by running a query that should fail if the SKU under 1400 (image save command). Since we could also fail due to the server not being instantiated, we use a backoff decorator to retry the command a few times before giving up.

Raises:

Type Description
TypeError

If the SSAS instance is running an incorrect SKU version.
ValueError

If the SSAS instance is not running or the command fails for another reason. This is used to trigger the backoff retry.
Source code in pbi_core/ssas/server/server.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
@backoff.on_exception(backoff.expo, ValueError, max_time=10)
def check_ssas_sku(self) -> None:
    """Checks if the SSAS instance is running the correct SKU version.

    Tests this assumption by running a query that should fail if the SKU under 1400 (image save command).
    Since we could also fail due to the server not being instantiated, we use a backoff decorator
    to retry the command a few times before giving up.

    Raises:
        TypeError: If the SSAS instance is running an incorrect SKU version.
        ValueError: If the SSAS instance is not running or the command fails for another reason.
            This is used to trigger the backoff retry.

    """
    try:
        self.query_xml(
            COMMAND_TEMPLATES["image_save.xml"].render(
                target_path="---",
                db_name="---",
            ),  # specifically choosing non-existant values to verify we get at least one error
        )
    except pbi_pyadomd.AdomdErrorResponseException as e:
        error_message = str(e.Message)
        if error_message == SKU_ERROR:
            return
        msg = f"Incorrect SKUVersion. We got the error: {error_message}"
        raise TypeError(msg) from None
    msg = "Got a 'file not loaded' type error. Waiting"
    raise ValueError(msg)

conn #

conn(db_name: str | None = None) -> pbi_pyadomd.Connection

Returns a pbi_pyadomd connection.

Source code in pbi_core/ssas/server/server.py 
51
52
53
def conn(self, db_name: str | None = None) -> pbi_pyadomd.Connection:
    """Returns a pbi_pyadomd connection."""
    return pbi_pyadomd.connect(self.conn_str(db_name))

conn_str #

conn_str(db_name: str | None = None) -> str

Formats the connection string for connecting to the background SSAS instance.

Source code in pbi_core/ssas/server/server.py 
45
46
47
48
49
def conn_str(self, db_name: str | None = None) -> str:
    """Formats the connection string for connecting to the background SSAS instance."""
    if db_name:
        return f"Provider=MSOLAP;Data Source={self.host}:{self.port};Initial Catalog={db_name};"
    return f"Provider=MSOLAP;Data Source={self.host}:{self.port};"

query_dax #

query_dax(query: str, *, db_name: str | bool | None = True) -> list[dict[str, Any]]

db_name: when bool and == True, uses the DB last loaded by this server instance.

(almost always the db of the loaded PBI unless you're manually reassigning server instances) when None or False, no db is supplied when a string, just passed to the client

Source code in pbi_core/ssas/server/server.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
def query_dax(self, query: str, *, db_name: str | bool | None = True) -> list[dict[str, Any]]:
    """db_name: when bool and == True, uses the DB last loaded by this server instance.

    (almost always the db of the loaded PBI unless you're manually reassigning server instances)
    when None or False, no db is supplied
    when a string, just passed to the client
    """
    if db_name is True:
        db_name = self.default_db
    elif db_name is False:
        db_name = None
    with self.conn(db_name) as conn:
        reader = conn.execute_dax(query)
        return reader.fetch_many()

query_xml #

query_xml(query: str, db_name: str | None = None) -> BeautifulSoup

Submits an XMLA query to the SSAS instance and returns the result as a BeautifulSoup object.

The query should be a valid XMLA command.

Parameters:

Name Type Description Default
query str

The XMLA query to execute.

 required
db_name str | None

The name of the database to execute the query against.

 None

Returns:

Name Type Description
BeautifulSoup BeautifulSoup

The result of the query parsed as a BeautifulSoup object.
Source code in pbi_core/ssas/server/server.py 
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
def query_xml(self, query: str, db_name: str | None = None) -> BeautifulSoup:
    """Submits an XMLA query to the SSAS instance and returns the result as a BeautifulSoup object.

    The query should be a valid XMLA command.

    Args:
        query (str): The XMLA query to execute.
        db_name (str | None): The name of the database to execute the query against.

    Returns:
        BeautifulSoup: The result of the query parsed as a BeautifulSoup object.

    """
    with self.conn(db_name) as conn:
        return conn.execute_xml(query)

remove_invalid_db_name_chars staticmethod #

remove_invalid_db_name_chars(orig_db_name: str) -> str

Utility function to convert a PBIX report name to an equivalent name for the DB in the SSAS instance.

Note

Raises a warning if the db_name is changed to inform user that the db_name does not match their input

Source code in pbi_core/ssas/server/server.py 
135
136
137
138
139
140
141
142
143
144
145
146
147
@staticmethod
def remove_invalid_db_name_chars(orig_db_name: str) -> str:
    """Utility function to convert a PBIX report name to an equivalent name for the DB in the SSAS instance.

    Note:
        Raises a warning if the db_name is changed to inform user that the db_name does not match their input

    """
    db_name = orig_db_name.replace("&", " ")[:100]
    db_name = db_name.strip()  # needed to find the correct name, since SSAS does stripping too
    if orig_db_name != db_name:
        logger.warning("db_name changed", original_name=orig_db_name, new_name=db_name)
    return db_name

sanitize_xml staticmethod #

sanitize_xml(xml_text: str) -> str

Method to XML-encode characters like "&" so that the Adomd connection doesn't mangle the XMLA commands.

Source code in pbi_core/ssas/server/server.py 
130
131
132
133
@staticmethod
def sanitize_xml(xml_text: str) -> str:
    """Method to XML-encode characters like "&" so that the Adomd connection doesn't mangle the XMLA commands."""
    return xml_text.replace("&", "&")

tabular_models #

tabular_models() -> list[BaseTabularModel]

Creates a list of the Tabular models existing in the SSAS server.

Note

Generally tabular models in the local environment correspond 1-1 with a PBIX report open in the Desktop app

Source code in pbi_core/ssas/server/server.py 
89
90
91
92
93
94
95
96
97
98
def tabular_models(self) -> list[BaseTabularModel]:
    """Creates a list of the Tabular models existing in the SSAS server.

    Note:
        Generally tabular models in the local environment correspond 1-1 with a PBIX report open in the Desktop app

    """
    # Query based on https://learn.microsoft.com/en-us/previous-versions/sql/sql-server-2012/ms126314(v=sql.110)
    dbs = self.query_dax(COMMAND_TEMPLATES["list_dbs.xml"].render())
    return [BaseTabularModel(row["CATALOG_NAME"], self) for row in dbs]