Skip to content

Table#

Bases: SsasRefreshRecord

TBD.

SSAS spec: Microsoft

Source code in pbi_core/ssas/model_tables/table/table.py
@define()
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)
    """

    _default_refresh_type: RefreshType = field(default=RefreshType.DATA_ONLY, init=False, repr=False, eq=False)

    alternate_source_precedence: int = field(eq=True)
    calculation_group_id: int | None = field(default=None, eq=True)
    data_category: DataCategory | None = field(default=None, eq=True)
    default_detail_rows_defintion_id: int | None = field(default=None, eq=True)
    description: str | None = field(default=None, eq=True)
    """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 = field(default=False, eq=True)
    """Controls whether this table is included in the model-wide refresh process"""
    is_hidden: bool = field(default=False, eq=True)
    """Controls whether the table appears in the edit mode of the report"""
    is_private: bool = field(default=False, eq=True)
    model_id: int = field(eq=True, repr=False)
    """The ID of the model this table belongs to"""
    name: str = field(eq=True)
    """The name of the table as it appears in the report"""
    refresh_policy_id: int | None = field(default=None, eq=True)
    show_as_variations_only: bool = field(default=False, eq=True)
    system_flags: int = field(eq=True)
    system_managed: bool | None = field(default=None, eq=True)
    table_storage_id: int | None = field(default=None, eq=True)

    lineage_tag: UUID = field(factory=uuid4, eq=True, repr=False)
    source_lineage_tag: UUID = field(factory=uuid4, eq=True, repr=False)

    modified_time: Final[datetime.datetime] = field(eq=False, on_setattr=setters.frozen, repr=False)
    structure_modified_time: Final[datetime.datetime] = field(eq=False, on_setattr=setters.frozen, repr=False)

    _commands: RefreshCommands = field(default=SsasCommands.table, init=False, repr=False, eq=False)

    def set_name(self, new_name: str, layout: "Layout") -> None:
        """Renames the measure and update any dependent expressions to use the new name.

        Since measures are referenced by name in DAX expressions, renaming a measure will break any dependent
        expressions.
        """
        entities = layout.find_all(Entity, lambda e: e.Entity == self.name)
        for entity in entities:
            entity.Entity = new_name

        set_name.fix_dax(self, new_name)
        self.name = new_name

    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.

        These are the measures that can be found under the table in the model structure.

        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 hierarchies(self) -> set["Hierarchy"]:
        """Get associated dependent hierarchies.

        Returns:
            (set[Hierarchy]): A list of the hierarchies defined on this table

        """
        return self._tabular_model.hierarchies.find_all({"table_id": self.id})

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

    def children_base(self) -> frozenset["LinkedEntity"]:
        # We don't include the table_measures since they are picked up via the columns' child measures
        return (
            LinkedEntity.from_iter(self.annotations(), by="annotation")
            | LinkedEntity.from_iter(
                self.columns(),
                by="column",
            )
            | LinkedEntity.from_iter(
                self.partitions(),
                by="partition",
            )
            | LinkedEntity.from_iter(
                self.measures(),
                by="measure",
            )
            | LinkedEntity.from_iter(
                self.perspective_tables(),
                by="perspective_table",
            )
            | LinkedEntity.from_iter(
                self.table_measures(),
                by="table_measure",
            )
            | LinkedEntity.from_iter(
                self.hierarchies(),
                by="hierarchy",
            )
            | LinkedEntity.from_iter(
                {self.refresh_policy()},
                by="refresh_policy",
            )
            | LinkedEntity.from_iter(
                {self.default_row_definition()},
                by="default_row_definition",
            )
            | LinkedEntity.from_iter(
                {self.calculation_group()},
                by="calculation_group",
            )
        )

    def perspective_tables(self) -> set["PerspectiveTable"]:
        return self._tabular_model.perspective_tables.find_all({"table_id": self.id})

    def parents_base(self) -> frozenset["LinkedEntity"]:
        return LinkedEntity.from_iter({self.model()}, by="model")

    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()]

    def external_sources(self) -> list["BaseExternalSource"]:
        return list({source for partition in self.partitions() for source in partition.external_sources()})

description class-attribute instance-attribute #

description: str | None = field(default=None, eq=True)

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

exclude_from_model_refresh class-attribute instance-attribute #

exclude_from_model_refresh: bool = field(default=False, eq=True)

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

is_hidden class-attribute instance-attribute #

is_hidden: bool = field(default=False, eq=True)

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

model_id class-attribute instance-attribute #

model_id: int = field(eq=True, repr=False)

The ID of the model this table belongs to

name class-attribute instance-attribute #

name: str = field(eq=True)

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
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
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,
    )

hierarchies #

hierarchies() -> set[Hierarchy]

Get associated dependent hierarchies.

Returns:

Type Description
set[Hierarchy]

A list of the hierarchies defined on this table

Source code in pbi_core/ssas/model_tables/table/table.py
def hierarchies(self) -> set["Hierarchy"]:
    """Get associated dependent hierarchies.

    Returns:
        (set[Hierarchy]): A list of the hierarchies defined on this table

    """
    return self._tabular_model.hierarchies.find_all({"table_id": self.id})

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
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
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
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()]

set_name #

set_name(new_name: str, layout: Layout) -> None

Renames the measure and update any dependent expressions to use the new name.

Since measures are referenced by name in DAX expressions, renaming a measure will break any dependent expressions.

Source code in pbi_core/ssas/model_tables/table/table.py
def set_name(self, new_name: str, layout: "Layout") -> None:
    """Renames the measure and update any dependent expressions to use the new name.

    Since measures are referenced by name in DAX expressions, renaming a measure will break any dependent
    expressions.
    """
    entities = layout.find_all(Entity, lambda e: e.Entity == self.name)
    for entity in entities:
        entity.Entity = new_name

    set_name.fix_dax(self, new_name)
    self.name = new_name

table_measures #

table_measures() -> set[Measure]

Get measures saved to this table.

These are the measures that can be found under the table in the model structure.

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
def table_measures(self) -> set[Measure]:
    """Get measures saved to this table.

    These are the measures that can be found under the table in the model structure.

    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})