Skip to content

Reader

Source code in pbi_pyadomd/reader.py 
 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
148
149
150
151
152
153
154
155
156
class Reader:
    _reader: "IDataReader"

    def __init__(self, reader: "IDataReader") -> None:
        self._reader = reader

    def __enter__(self) -> "Reader":
        return self

    def __exit__(
        self,
        exc_type: type[BaseException] | None,
        exc_value: BaseException | None,
        traceback: "TracebackType | None",  # noqa: PYI036
    ) -> None:
        self.close()

    def close(self) -> None:
        self._reader.Close()

    def column_names(self) -> list[str]:
        """Returns the column names of the last executed query."""
        return [self._reader.GetName(i) for i in range(self.field_count)]

    def descriptions(self) -> list[Description]:
        return [
            Description(
                self._reader.GetName(i),
                adomd_type_map[self._reader.GetFieldType(i).ToString()].type_name,
            )
            for i in range(self.field_count)
        ]

    def fetch_one(self) -> dict[str, Any]:
        """Fetches a single row from the last executed query as a dictionary.

        Returns
        -------
            dict[str, Any]: A dictionary representing the row, with column names as keys

        """
        column_names = self.column_names()
        data = self.fetch_one_tuple()
        return dict(zip(column_names, data, strict=False))

    def fetch_one_tuple(self) -> tuple[Any, ...]:
        """Fetches a single row from the last executed query as a tuple.

        Returns:
            tuple[Any, ...]: A tuple representing the row, with C# values converted
                to their appropriate python types.

        """
        return tuple(
            convert(
                self._reader.GetFieldType(i).ToString(),
                self._reader[i],
                adomd_type_map,
            )
            for i in range(self.field_count)
        )

    def fetch_many(self, limit: int | None = None) -> list[dict[str, Any]]:
        """Fetches multiple rows from the last executed query.

        Args:
            limit (int | None): The number of rows to fetch. If None, fetches all rows.

        Returns:
            list[dict[str, Any]]: A list of dictionaries representing the rows.

        """
        if limit is not None:
            return [self.fetch_one() for _ in range(limit) if self.read()]
        return list(self.fetch_stream())

    def fetch_stream(self) -> Iterator[dict[str, Any]]:
        """Fetches the rows from the last executed query as a stream of dictionaries.

        Note:
        ----
            You may need to close the reader after fetching the rows if:

            1. You are using a explicit limit that is shorter than the total number of
            rows in the query result
            2. You are tracing the command associated with the reader

            This is because the trace will not create a query end record (since it
            assumes the client is still reading) without explicitly closing the
            reader. The reader can be closed with `self._reader.Close()`

        """
        column_names = self.column_names()
        while self.read():
            yield dict(zip(column_names, self.fetch_one_tuple(), strict=False))

    @property
    def field_count(self) -> int:
        return self._reader.FieldCount

    @property
    def is_closed(self) -> bool:
        return self._reader.IsClosed

    def read(self) -> bool:
        try:
            return self._reader.Read()
        except AdomdUnknownResponseException:
            return False

    def read_outer_xml(self) -> str:
        return self._reader.ReadOuterXml()

column_names 

column_names() -> list[str]

Returns the column names of the last executed query.

Source code in pbi_pyadomd/reader.py 
65
66
67
def column_names(self) -> list[str]:
    """Returns the column names of the last executed query."""
    return [self._reader.GetName(i) for i in range(self.field_count)]

fetch_many 

fetch_many(limit: int | None = None) -> list[dict[str, Any]]

Fetches multiple rows from the last executed query.

Parameters:

Name Type Description Default
limit int | None

The number of rows to fetch. If None, fetches all rows.

 None

Returns:

Type Description
list[dict[str, Any]]

list[dict[str, Any]]: A list of dictionaries representing the rows.
Source code in pbi_pyadomd/reader.py 
107
108
109
110
111
112
113
114
115
116
117
118
119
def fetch_many(self, limit: int | None = None) -> list[dict[str, Any]]:
    """Fetches multiple rows from the last executed query.

    Args:
        limit (int | None): The number of rows to fetch. If None, fetches all rows.

    Returns:
        list[dict[str, Any]]: A list of dictionaries representing the rows.

    """
    if limit is not None:
        return [self.fetch_one() for _ in range(limit) if self.read()]
    return list(self.fetch_stream())

fetch_one 

fetch_one() -> dict[str, Any]

Fetches a single row from the last executed query as a dictionary.

Returns

dict[str, Any]: A dictionary representing the row, with column names as keys
Source code in pbi_pyadomd/reader.py 
78
79
80
81
82
83
84
85
86
87
88
def fetch_one(self) -> dict[str, Any]:
    """Fetches a single row from the last executed query as a dictionary.

    Returns
    -------
        dict[str, Any]: A dictionary representing the row, with column names as keys

    """
    column_names = self.column_names()
    data = self.fetch_one_tuple()
    return dict(zip(column_names, data, strict=False))

fetch_one_tuple 

fetch_one_tuple() -> tuple[Any, ...]

Fetches a single row from the last executed query as a tuple.

Returns:

Type Description
tuple[Any, ...]

tuple[Any, ...]: A tuple representing the row, with C# values converted to their appropriate python types.
Source code in pbi_pyadomd/reader.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
def fetch_one_tuple(self) -> tuple[Any, ...]:
    """Fetches a single row from the last executed query as a tuple.

    Returns:
        tuple[Any, ...]: A tuple representing the row, with C# values converted
            to their appropriate python types.

    """
    return tuple(
        convert(
            self._reader.GetFieldType(i).ToString(),
            self._reader[i],
            adomd_type_map,
        )
        for i in range(self.field_count)
    )

fetch_stream 

fetch_stream() -> Iterator[dict[str, Any]]

Fetches the rows from the last executed query as a stream of dictionaries.

Note:

You may need to close the reader after fetching the rows if:

1. You are using a explicit limit that is shorter than the total number of
rows in the query result
2. You are tracing the command associated with the reader

This is because the trace will not create a query end record (since it
assumes the client is still reading) without explicitly closing the
reader. The reader can be closed with `self._reader.Close()`
Source code in pbi_pyadomd/reader.py 
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
def fetch_stream(self) -> Iterator[dict[str, Any]]:
    """Fetches the rows from the last executed query as a stream of dictionaries.

    Note:
    ----
        You may need to close the reader after fetching the rows if:

        1. You are using a explicit limit that is shorter than the total number of
        rows in the query result
        2. You are tracing the command associated with the reader

        This is because the trace will not create a query end record (since it
        assumes the client is still reading) without explicitly closing the
        reader. The reader can be closed with `self._reader.Close()`

    """
    column_names = self.column_names()
    while self.read():
        yield dict(zip(column_names, self.fetch_one_tuple(), strict=False))