new file: __pycache__/_type_code.cpython-313.pyc

new file:   __pycache__/config.cpython-313.pyc
	new file:   _type_code.py
	new file:   config.py
	new file:   connectAS.py
	new file:   msnet/Microsoft.AnalysisServices.AdomdClient.dll
	new file:   msnet/Microsoft.AnalysisServices.DLL
	new file:   msnet/Microsoft.AnalysisServices.Tabular.DLL
	new file:   test.ipynb

_type_code.py

@@ -0,0 +1,39 @@
+from typing import Any, cast, Callable, Dict, Iterator, List, NamedTuple, Optional, Tuple, TypeVar
+from System import Decimal
+from datetime import datetime
+from functools import partial
+
+
+#Types
+F = Callable[[Any], Any]
+class Type_code(NamedTuple):
+    type_obj:F
+    type_name:str
+
+def _option_type(datatype, data):
+    if data:
+        return datatype(data)
+    if datatype in [bool, int, float] and data == 0:
+        return datatype(data)
+    return None
+
+adomd_type_map:Dict[str, Type_code] = {
+    'System.Boolean': Type_code(partial(_option_type, bool), bool.__name__),
+    'System.DateTime': Type_code(lambda x: datetime(x.Year, x.Month, x.Day, x.Hour, x.Minute, x.Second) if x else None, datetime.__name__),
+    'System.Decimal': Type_code(lambda x: Decimal.ToDouble(x) if x else None, float.__name__),
+    'System.Double': Type_code(partial(_option_type, float), float.__name__),
+    'System.Single': Type_code(partial(_option_type, float), float.__name__),
+    'System.String': Type_code(partial(_option_type, str), str.__name__),
+    'System.Guid': Type_code(partial(_option_type, str), str.__name__),
+    'System.UInt16': Type_code(partial(_option_type, int), int.__name__),
+    'System.UInt32': Type_code(partial(_option_type, int), int.__name__),
+    'System.UInt64': Type_code(partial(_option_type, int), int.__name__),
+    'System.Int16': Type_code(partial(_option_type, int), int.__name__),
+    'System.Int32':  Type_code(partial(_option_type, int), int.__name__),
+    'System.Int64': Type_code(partial(_option_type, int), int.__name__),
+    'System.Object': Type_code(lambda x: x, 'System.Object'),
+}
+
+def convert(datatype:str, data:Any, type_map:Dict[str, Type_code]):
+    type_to_convert = type_map[datatype]
+    return type_to_convert.type_obj(data)

config.py

@@ -0,0 +1,6 @@
+import os
+
+# 获取当前文件所在的文件夹路径
+CURRENT_DIR_PATH = os.path.dirname(os.path.abspath(__file__))
+# MS.NET包的路径
+NET_FOLDER = "msnet"

connectAS.py

@@ -0,0 +1,270 @@
+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}")