Local SSAS Server

Bases: BaseServer

A Server running locally on the user's machine.

This subclass has the ability to load/dump Tabular Models to a PBIX file. Also creates a background SSAS instance and workspace to handle processing if none is provided.

Parameters:

Name	Type	Description	Default
kill_on_exit	bool	Indicates if the background SSAS instance handling processing should be terminated at the end of the python session	True

Source code in pbi_core/ssas/server/server.py

class LocalServer(BaseServer):
    """A Server running locally on the user's machine.

    This subclass has the ability to load/dump Tabular Models
    to a PBIX file. Also creates a background SSAS instance and workspace to handle processing if none is provided.

    Args:
        kill_on_exit (bool): Indicates if the background SSAS instance handling
            processing should be terminated at the end of the python session

    """

    physical_process: SSASProcess
    """
    A Python class handling the lifetime of the SSAS Instance. Interacts with the SSAS instance only as a process
    """

    def __init__(
        self,
        host: str = "localhost",
        workspace_directory: "StrPath | None" = None,
        pid: int | None = None,
        *,
        kill_on_exit: bool = True,
    ) -> None:
        if pid is not None:
            self.physical_process = SSASProcess(pid=pid, kill_on_exit=kill_on_exit, startup_config=None)
        else:
            startup_config = get_startup_config()
            workspace_directory = workspace_directory or (
                startup_config.workspace_dir / datetime.now(UTC).strftime(DT_FORMAT)
            )
            self.physical_process = SSASProcess(
                workspace_directory=workspace_directory,
                kill_on_exit=kill_on_exit,
                startup_config=startup_config,
            )
        super().__init__(host, self.physical_process._port)

    def load_pbix(self, path: "StrPath", *, db_name: str | None = None) -> LocalTabularModel:
        """Takes a Path to a PBIX report and loads the PBIX Datamodel to the SSAS instance in the SSASProcess.

        Raises:
            FileNotFoundError: when the path to the PBIX file does not exist
            AdomdErrorResponseException: Occurs when the DB already exists

        """
        path = pathlib.Path(path)
        if not path.exists():
            msg = f"The path to the PBIX does not exist: {path.absolute().as_posix()}"
            raise FileNotFoundError(msg)
        if db_name is None:
            db_name = path.stem
        db_name = self.remove_invalid_db_name_chars(db_name)
        load_command = COMMAND_TEMPLATES["image_load.xml"].render(
            db_name=db_name,
            source_path=self.sanitize_xml(path.absolute().as_posix()),
        )
        try:
            self.query_xml(load_command)
        except pbi_pyadomd.AdomdErrorResponseException as e:
            if (
                "user does not have permission to restore the database, or the database already exists and AllowOverwrite is not specified"  # noqa: E501
                in str(e.Message)
            ):
                logger.warning("Removing old version of PBIX data model for new version", db_name=db_name)
                self.query_xml(COMMAND_TEMPLATES["db_delete.xml"].render(db_name=db_name))
                self.query_xml(load_command)
            else:
                raise pbi_pyadomd.AdomdErrorResponseException from e

        self.default_db = db_name  # needed so the DAX queries are pointed to the right DB by defauilt
        logger.info("Tabular Model load complete")
        tab_model = LocalTabularModel(db_name=db_name, server=self, pbix_path=path)
        tab_model.sync_from()
        return tab_model

    def save_pbix(self, path: "StrPath", db_name: str) -> None:
        path = pathlib.Path(path)
        self.query_xml(
            COMMAND_TEMPLATES["image_save.xml"].render(
                db_name=db_name,
                target_path=self.sanitize_xml(path.absolute().as_posix()),
            ),
        )

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

physical_process instance-attribute

physical_process: SSASProcess

A Python class handling the lifetime of the SSAS Instance. Interacts with the SSAS instance only as a process

load_pbix

load_pbix(path: StrPath, *, db_name: str | None = None) -> LocalTabularModel

Takes a Path to a PBIX report and loads the PBIX Datamodel to the SSAS instance in the SSASProcess.

Raises:

Type	Description
FileNotFoundError	when the path to the PBIX file does not exist
AdomdErrorResponseException	Occurs when the DB already exists

Source code in pbi_core/ssas/server/server.py

def load_pbix(self, path: "StrPath", *, db_name: str | None = None) -> LocalTabularModel:
    """Takes a Path to a PBIX report and loads the PBIX Datamodel to the SSAS instance in the SSASProcess.

    Raises:
        FileNotFoundError: when the path to the PBIX file does not exist
        AdomdErrorResponseException: Occurs when the DB already exists

    """
    path = pathlib.Path(path)
    if not path.exists():
        msg = f"The path to the PBIX does not exist: {path.absolute().as_posix()}"
        raise FileNotFoundError(msg)
    if db_name is None:
        db_name = path.stem
    db_name = self.remove_invalid_db_name_chars(db_name)
    load_command = COMMAND_TEMPLATES["image_load.xml"].render(
        db_name=db_name,
        source_path=self.sanitize_xml(path.absolute().as_posix()),
    )
    try:
        self.query_xml(load_command)
    except pbi_pyadomd.AdomdErrorResponseException as e:
        if (
            "user does not have permission to restore the database, or the database already exists and AllowOverwrite is not specified"  # noqa: E501
            in str(e.Message)
        ):
            logger.warning("Removing old version of PBIX data model for new version", db_name=db_name)
            self.query_xml(COMMAND_TEMPLATES["db_delete.xml"].render(db_name=db_name))
            self.query_xml(load_command)
        else:
            raise pbi_pyadomd.AdomdErrorResponseException from e

    self.default_db = db_name  # needed so the DAX queries are pointed to the right DB by defauilt
    logger.info("Tabular Model load complete")
    tab_model = LocalTabularModel(db_name=db_name, server=self, pbix_path=path)
    tab_model.sync_from()
    return tab_model