Skip to content

Table#

Bases: SsasRefreshRecord

TBD.

SSAS spec: Microsoft

Source code in pbi_core/ssas/model_tables/table/table.py 
 24
 25
 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
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
class Table(SsasRefreshRecord):
    """TBD.

    SSAS spec: [Microsoft](https://learn.microsoft.com/en-us/openspecs/sql_server_protocols/ms-ssas-t/6360ac84-0717-4170-bce0-284cbef419ca)
    """

    _refresh_type = RefreshType.DATA_ONLY

    alternate_source_precedence: int
    calculation_group_id: int | None = None
    data_category: DataCategory | None = None
    default_detail_rows_defintion_id: int | None = None
    description: str | None = None
    """A description of the table, which may be used in the hover tooltip in edit mode"""
    exclude_from_automatic_aggregations: bool = False
    exclude_from_model_refresh: bool
    """Controls whether this table is included in the model-wide refresh process"""
    is_hidden: bool
    """Controls whether the table appears in the edit mode of the report"""
    is_private: bool
    model_id: int
    """The ID of the model this table belongs to"""
    name: str
    """The name of the table as it appears in the report"""
    refresh_policy_id: int | None = None
    show_as_variations_only: bool
    system_flags: int
    system_managed: bool | None = None
    table_storage_id: int | None = None

    lineage_tag: UUID = uuid4()
    source_lineage_tag: UUID = uuid4()

    modified_time: datetime.datetime
    structure_modified_time: datetime.datetime

    _commands: RefreshCommands = PrivateAttr(default_factory=lambda: SsasCommands.table)

    def modification_hash(self) -> int:
        return hash((
            self.name,
            self.description,
            self.is_hidden,
        ))

    def calculation_group(self) -> "CalculationGroup | None":
        if self.calculation_group_id is None:
            return None
        return self.tabular_model.calculation_groups.find(self.calculation_group_id)

    def refresh_policy(self) -> "RefreshPolicy | None":
        if self.refresh_policy_id is None:
            return None
        return self.tabular_model.refresh_policies.find(self.refresh_policy_id)

    def is_system_table(self) -> bool:
        return bool(self.system_flags >> 1 % 2)

    def is_from_calculated_table(self) -> bool:
        return bool(self.system_flags % 2)

    def data(self, head: int = 100) -> list[dict[str, int | float | str]]:
        """Extracts records from the table in SSAS.

        Args:
            head (int): The number of records to return from the table.

        Returns:
            list[dict[str, int | float | str]]: A list of SSAS records in dictionary form.
                The keys are the field names and the values are the record values

        """
        return self.tabular_model.server.query_dax(
            f"EVALUATE TOPN({head}, ALL('{self.name}'))",
            db_name=self.tabular_model.db_name,
        )

    def partitions(self) -> set[Partition]:
        """Get associated dependent partitions.

        Returns:
            (set[Partition]): A list of the partitions containing data for this table

        """
        return self.tabular_model.partitions.find_all({"table_id": self.id})

    def columns(self) -> set[Column]:
        """Get associated dependent partitions.

        Returns:
            (set[Column]): A list of the columns in this table

        """
        return self.tabular_model.columns.find_all({"table_id": self.id})

    def default_row_definition(self) -> "DetailRowDefinition | None":
        if self.default_detail_rows_defintion_id is None:
            return None
        return self.tabular_model.detail_row_definitions.find(self.default_detail_rows_defintion_id)

    def table_measures(self) -> set[Measure]:
        """Get measures saved to this table.

        Returns:
            (set[Measure]): A list of measures saved to this table

        Note:
            These measures do not necessarily have calculations that depend on this table.
                For that use `table.measures()`

        """
        return self.tabular_model.measures.find_all({"table_id": self.id})

    def measures(self, *, recursive: bool = False) -> set[Measure]:
        """Get measures that logically depend on this table.

        Examples:
            ```python
            print(measure.expression)
            # sumx(example, [a])

            Table(name=example).measures()
            # [..., Measure(name='measure'), ...]
            ```
        Args:
            recursive (bool): Whether to include measures that depend on other measures.

        Returns:
            (set[Measure]): A list of measures that logically depend this table

        Note:
            These measures are not necessarily saved physically to this table. For that use `table.table_measures()`

        """
        ret = set()
        for col in self.columns():
            ret.update(col.child_measures(recursive=recursive))
        return ret

    def model(self) -> "Model":
        return self.tabular_model.model

    def get_lineage(self, lineage_type: Literal["children", "parents"]) -> LineageNode:
        if lineage_type == "children":
            return LineageNode(
                self,
                lineage_type,
                [col.get_lineage(lineage_type) for col in self.columns()]
                + [partition.get_lineage(lineage_type) for partition in self.partitions()]
                + [measure.get_lineage(lineage_type) for measure in self.measures()],
            )
        return LineageNode(self, lineage_type, [self.model().get_lineage(lineage_type)])

    def refresh(self, *, include_model_refresh: bool = True) -> list[BeautifulSoup]:  # pyright: ignore reportIncompatibleMethodOverride
        """Needs a model refresh to properly propogate the update."""
        if include_model_refresh:
            return [
                super().refresh(),
                self.model().refresh(),
            ]
        return [super().refresh()]

description class-attribute instance-attribute #

description: str | None = None

A description of the table, which may be used in the hover tooltip in edit mode

exclude_from_model_refresh instance-attribute #

exclude_from_model_refresh: bool

Controls whether this table is included in the model-wide refresh process

is_hidden instance-attribute #

is_hidden: bool

Controls whether the table appears in the edit mode of the report

model_id instance-attribute #

model_id: int

The ID of the model this table belongs to

name instance-attribute #

name: str

The name of the table as it appears in the report

columns #

columns() -> set[Column]

Get associated dependent partitions.

Returns:

Type Description
set[Column]

A list of the columns in this table
Source code in pbi_core/ssas/model_tables/table/table.py 
110
111
112
113
114
115
116
117
def columns(self) -> set[Column]:
    """Get associated dependent partitions.

    Returns:
        (set[Column]): A list of the columns in this table

    """
    return self.tabular_model.columns.find_all({"table_id": self.id})

data #

data(head: int = 100) -> list[dict[str, int | float | str]]

Extracts records from the table in SSAS.

Parameters:

Name Type Description Default
head int

The number of records to return from the table.

 100

Returns:

Type Description
list[dict[str, int | float | str]]

list[dict[str, int | float | str]]: A list of SSAS records in dictionary form. The keys are the field names and the values are the record values
Source code in pbi_core/ssas/model_tables/table/table.py 
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
def data(self, head: int = 100) -> list[dict[str, int | float | str]]:
    """Extracts records from the table in SSAS.

    Args:
        head (int): The number of records to return from the table.

    Returns:
        list[dict[str, int | float | str]]: A list of SSAS records in dictionary form.
            The keys are the field names and the values are the record values

    """
    return self.tabular_model.server.query_dax(
        f"EVALUATE TOPN({head}, ALL('{self.name}'))",
        db_name=self.tabular_model.db_name,
    )

measures #

measures(*, recursive: bool = False) -> set[Measure]

Get measures that logically depend on this table.

Examples:

print(measure.expression)
# sumx(example, [a])

Table(name=example).measures()
# [..., Measure(name='measure'), ...]

Args: recursive (bool): Whether to include measures that depend on other measures.

Returns:

Type Description
set[Measure]

A list of measures that logically depend this table
Note

These measures are not necessarily saved physically to this table. For that use table.table_measures()

Source code in pbi_core/ssas/model_tables/table/table.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
def measures(self, *, recursive: bool = False) -> set[Measure]:
    """Get measures that logically depend on this table.

    Examples:
        ```python
        print(measure.expression)
        # sumx(example, [a])

        Table(name=example).measures()
        # [..., Measure(name='measure'), ...]
        ```
    Args:
        recursive (bool): Whether to include measures that depend on other measures.

    Returns:
        (set[Measure]): A list of measures that logically depend this table

    Note:
        These measures are not necessarily saved physically to this table. For that use `table.table_measures()`

    """
    ret = set()
    for col in self.columns():
        ret.update(col.child_measures(recursive=recursive))
    return ret

partitions #

partitions() -> set[Partition]

Get associated dependent partitions.

Returns:

Type Description
set[Partition]

A list of the partitions containing data for this table
Source code in pbi_core/ssas/model_tables/table/table.py 
101
102
103
104
105
106
107
108
def partitions(self) -> set[Partition]:
    """Get associated dependent partitions.

    Returns:
        (set[Partition]): A list of the partitions containing data for this table

    """
    return self.tabular_model.partitions.find_all({"table_id": self.id})

refresh #

refresh(*, include_model_refresh: bool = True) -> list[BeautifulSoup]

Needs a model refresh to properly propogate the update.

Source code in pbi_core/ssas/model_tables/table/table.py 
177
178
179
180
181
182
183
184
def refresh(self, *, include_model_refresh: bool = True) -> list[BeautifulSoup]:  # pyright: ignore reportIncompatibleMethodOverride
    """Needs a model refresh to properly propogate the update."""
    if include_model_refresh:
        return [
            super().refresh(),
            self.model().refresh(),
        ]
    return [super().refresh()]

table_measures #

table_measures() -> set[Measure]

Get measures saved to this table.

Returns:

Type Description
set[Measure]

A list of measures saved to this table
Note

These measures do not necessarily have calculations that depend on this table. For that use table.measures()

Source code in pbi_core/ssas/model_tables/table/table.py 
124
125
126
127
128
129
130
131
132
133
134
135
def table_measures(self) -> set[Measure]:
    """Get measures saved to this table.

    Returns:
        (set[Measure]): A list of measures saved to this table

    Note:
        These measures do not necessarily have calculations that depend on this table.
            For that use `table.measures()`

    """
    return self.tabular_model.measures.find_all({"table_id": self.id})