PyAnalysisService/connectAS.py

from __future__ import annotations

# from pyadomd import *


import clr
import os
from System.IO import FileNotFoundException
from config import CURRENT_DIR_PATH, NET_FOLDER

# MS.NET path
AnalysisServices_path = os.path.join(
    CURRENT_DIR_PATH, NET_FOLDER, "Microsoft.AnalysisServices"
)
AdomdClient_path = os.path.join(
    CURRENT_DIR_PATH, NET_FOLDER, "Microsoft.AnalysisServices.AdomdClient"
)
Tabular_path = os.path.join(
    CURRENT_DIR_PATH, NET_FOLDER, "Microsoft.AnalysisServices.Tabular"
)

# Add reference
try:
    clr.AddReference(AdomdClient_path)
    clr.AddReference(Tabular_path)
    from Microsoft.AnalysisServices.AdomdClient import AdomdConnection, AdomdCommand
    from Microsoft.AnalysisServices.Tabular import (
        Database,
        Server,
        Table,
        Column,
        Measure,
        Partition,
        Level,
        Hierarchy,
        Trace
    )
except FileNotFoundException as e:
    print(
        "========================================================================================"
    )
    print(e.ToString())
    print(
        "========================================================================================"
    )


from _type_code import adomd_type_map, convert, TypeVar, NamedTuple

# Types
T = TypeVar("T")


class Description(NamedTuple):
    """
    :param [name]: Column name
    :param [type_code]: The column data type
    """

    name: str
    type_code: str


class Cursor:

    def __init__(self, connection: AdomdConnection):
        self._conn = connection
        self._description: List[Description] = []

    def close(self) -> None:
        """
        Closes the cursor
        """
        if self.is_closed:
            return
        self._reader.Close()

    def execute(self, query: str) -> Cursor:
        """
        Executes a query against the data source

        :params [query]: The query to be executed
        """
        self._cmd = AdomdCommand(query, self._conn)
        self._reader = self._cmd.ExecuteReader()
        self._field_count = self._reader.FieldCount

        for i in range(self._field_count):
            self._description.append(
                Description(
                    self._reader.GetName(i),
                    adomd_type_map[self._reader.GetFieldType(i).ToString()].type_name,
                )
            )
        return self

    def fetchone(self) -> Iterator[Tuple[T, ...]]:
        """
        Fetches the current line from the last executed query
        """
        while self._reader.Read():
            yield tuple(
                convert(
                    self._reader.GetFieldType(i).ToString(),
                    self._reader[i],
                    adomd_type_map,
                )
                for i in range(self._field_count)
            )

    def fetchmany(self, size=1) -> List[Tuple[T, ...]]:
        """
        Fetches one or more lines from the last executed query

        :params [size]: The number of rows to fetch.
                        If the size parameter exceeds the number of rows returned from the last executed query then fetchmany will return all rows from that query.
        """
        l: List[Tuple[T, ...]] = []
        try:
            for i in range(size):
                l.append(next(self.fetchone()))
        except StopIteration:
            pass
        return l

    def fetchall(self) -> List[Tuple[T, ...]]:
        """
        Fetches all the rows from the last executed query
        """
        # mypy issues with list comprehension :-(
        return [i for i in self.fetchone()]  # type: ignore

    @property
    def is_closed(self) -> bool:
        try:
            state = self._reader.IsClosed
        except AttributeError:
            return True
        return state

    @property
    def description(self) -> List[Description]:
        return self._description

    def __enter__(self) -> Cursor:
        return self

    def __exit__(self, exc_type, exc_value, traceback) -> None:
        self.close()

class Pyadomd:

    def __init__(self, conn_str:str):
        self.conn = AdomdConnection()
        self.conn.ConnectionString = conn_str

    def close(self) -> None:
        """
        Closes the connection
        """
        self.conn.Close()
    
    def open(self) -> None:
        """
        Opens the connection
        """
        self.conn.Open()

    def cursor(self) -> Cursor:
        """
        Creates a cursor object
        """
        c = Cursor(self.conn)
        return c
    
    @property
    def state(self) -> int:
        """
        1 = Open
        0 = Closed
        """
        return self.conn.State
    
    def __enter__(self) -> Pyadomd:
        self.open()
        return self

    def __exit__(self, exc_type, exc_value, traceback) -> None:
        self.close()


class Pytabular:
    def __init__(self, conn_str: str):
        self.server = Server()
        self.server.Connect(conn_str)

    def close(self) -> None:
        """Closes the connection"""
        if self.server is not None:
            self.server.Disconnect()

    @property
    def databases(self) -> List[str]:
        """Returns a list of database names in the server"""
        return [db.Name for db in self.server.Databases]

    @property
    def state(self) -> int:
        """1 = Open, 0 = Closed"""
        return 1 if self.server is not None and self.server.Connected else 0

    def __enter__(self) -> Pytabular:
        return self

    def __exit__(self, exc_type, exc_value, traceback) -> None:
        self.close()

    # @property
    # def tables(self) -> List[Table]:
    #     """
    #     Returns a list of tables in the database
    #     """
    #     return self.db.Model.Tables

    # @property
    # def partitions(self) -> List[Partition]:
    #     """
    #     Returns a list of partitions in the database
    #     """
    #     return self.db.Model.Partitions

    # @property
    # def levels(self) -> List[Level]:
    #     """
    #     Returns a list of levels in the database
    #     """
    #     return self.db.Model.Levels

    # @property
    # def hierarchies(self) -> List[Hierarchy]:
    #     """
    #     Returns a list of hierarchies in the database
    #     """
    #     return self.db.Model.Hierarchies

    # @property
    # def measures(self) -> List[Measure]:
    #     """
    #     Returns a list of measures in the database
    #     """
    #     return self.db.Model.Measures

    # @property
    # def traces(self) -> List[Trace]:
    #     """
    #     Returns a list of traces in the database
    #     """
    #     return self.db.Model.Traces





if __name__ == "__main__":
    conn_str = "Provider=MSOLAP;Data Source=localhost"
    with Pytabular(conn_str) as conn:
        print("Connected to Analysis Services server")
        print("Available databases:")
        for db in conn.databases:
            print(f"- {db}")