Split classes in spec_classes.py into separate files.

metricflow-semantics/metricflow_semantics/specs/dimension_spec.py

@@ -0,0 +1,55 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Tuple
+
+from dbt_semantic_interfaces.dataclass_serialization import SerializableDataclass
+from dbt_semantic_interfaces.references import DimensionReference, EntityReference
+from typing_extensions import override
+
+from metricflow_semantics.model.semantics.linkable_element import ElementPathKey, LinkableElementType
+from metricflow_semantics.naming.linkable_spec_name import StructuredLinkableSpecName
+from metricflow_semantics.specs.instance_spec import InstanceSpecVisitor, LinkableInstanceSpec
+from metricflow_semantics.visitor import VisitorOutputT
+
+
+@dataclass(frozen=True)
+class DimensionSpec(LinkableInstanceSpec, SerializableDataclass):  # noqa: D101
+    element_name: str
+    entity_links: Tuple[EntityReference, ...]
+
+    @property
+    def without_first_entity_link(self) -> DimensionSpec:  # noqa: D102
+        assert len(self.entity_links) > 0, f"Spec does not have any entity links: {self}"
+        return DimensionSpec(element_name=self.element_name, entity_links=self.entity_links[1:])
+
+    @property
+    def without_entity_links(self) -> DimensionSpec:  # noqa: D102
+        return DimensionSpec(element_name=self.element_name, entity_links=())
+
+    @staticmethod
+    def from_linkable(spec: LinkableInstanceSpec) -> DimensionSpec:  # noqa: D102
+        return DimensionSpec(element_name=spec.element_name, entity_links=spec.entity_links)
+
+    @staticmethod
+    def from_name(name: str) -> DimensionSpec:
+        """Construct from a name e.g. listing__ds__month."""
+        parsed_name = StructuredLinkableSpecName.from_name(name)
+        return DimensionSpec(
+            entity_links=tuple([EntityReference(idl) for idl in parsed_name.entity_link_names]),
+            element_name=parsed_name.element_name,
+        )
+
+    @property
+    def reference(self) -> DimensionReference:  # noqa: D102
+        return DimensionReference(element_name=self.element_name)
+
+    def accept(self, visitor: InstanceSpecVisitor[VisitorOutputT]) -> VisitorOutputT:  # noqa: D102
+        return visitor.visit_dimension_spec(self)
+
+    @property
+    @override
+    def element_path_key(self) -> ElementPathKey:
+        return ElementPathKey(
+            element_name=self.element_name, element_type=LinkableElementType.DIMENSION, entity_links=self.entity_links
+        )

metricflow-semantics/metricflow_semantics/specs/entity_spec.py

@@ -0,0 +1,88 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any, Tuple
+
+from dbt_semantic_interfaces.dataclass_serialization import SerializableDataclass
+from dbt_semantic_interfaces.references import EntityReference
+from typing_extensions import override
+
+from metricflow_semantics.model.semantics.linkable_element import ElementPathKey, LinkableElementType
+from metricflow_semantics.naming.linkable_spec_name import StructuredLinkableSpecName
+from metricflow_semantics.specs.instance_spec import InstanceSpecVisitor, LinkableInstanceSpec
+from metricflow_semantics.visitor import VisitorOutputT
+
+
+@dataclass(frozen=True)
+class EntitySpec(LinkableInstanceSpec, SerializableDataclass):  # noqa: D101
+    @property
+    def without_first_entity_link(self) -> EntitySpec:  # noqa: D102
+        assert len(self.entity_links) > 0, f"Spec does not have any entity links: {self}"
+        return EntitySpec(element_name=self.element_name, entity_links=self.entity_links[1:])
+
+    @property
+    def without_entity_links(self) -> EntitySpec:  # noqa: D102
+        return LinklessEntitySpec.from_element_name(self.element_name)
+
+    @property
+    def as_linkless_prefix(self) -> Tuple[EntityReference, ...]:
+        """Creates tuple of linkless entities that could be included in the entity_links of another spec.
+
+        eg as a prefix to a DimensionSpec's entity links to when a join is occurring via this entity
+        """
+        return (EntityReference(element_name=self.element_name),) + self.entity_links
+
+    @staticmethod
+    def from_name(name: str) -> EntitySpec:  # noqa: D102
+        structured_name = StructuredLinkableSpecName.from_name(name)
+        return EntitySpec(
+            entity_links=tuple(EntityReference(idl) for idl in structured_name.entity_link_names),
+            element_name=structured_name.element_name,
+        )
+
+    def __eq__(self, other: Any) -> bool:  # type: ignore[misc]  # noqa: D105
+        if not isinstance(other, EntitySpec):
+            return False
+        return self.element_name == other.element_name and self.entity_links == other.entity_links
+
+    def __hash__(self) -> int:  # noqa: D105
+        return hash((self.element_name, self.entity_links))
+
+    @property
+    def reference(self) -> EntityReference:  # noqa: D102
+        return EntityReference(element_name=self.element_name)
+
+    def accept(self, visitor: InstanceSpecVisitor[VisitorOutputT]) -> VisitorOutputT:  # noqa: D102
+        return visitor.visit_entity_spec(self)
+
+    @property
+    @override
+    def element_path_key(self) -> ElementPathKey:
+        return ElementPathKey(
+            element_name=self.element_name, element_type=LinkableElementType.ENTITY, entity_links=self.entity_links
+        )
+
+
+@dataclass(frozen=True)
+class LinklessEntitySpec(EntitySpec, SerializableDataclass):
+    """Similar to EntitySpec, but requires that it doesn't have entity links."""
+
+    @staticmethod
+    def from_element_name(element_name: str) -> LinklessEntitySpec:  # noqa: D102
+        return LinklessEntitySpec(element_name=element_name, entity_links=())
+
+    def __post_init__(self) -> None:  # noqa: D105
+        if len(self.entity_links) > 0:
+            raise RuntimeError(f"{self.__class__.__name__} shouldn't have entity links. Got: {self}")
+
+    def __eq__(self, other: Any) -> bool:  # type: ignore[misc]  # noqa: D105
+        if not isinstance(other, EntitySpec):
+            return False
+        return self.element_name == other.element_name and self.entity_links == other.entity_links
+
+    def __hash__(self) -> int:  # noqa: D105
+        return hash((self.element_name, self.entity_links))
+
+    @staticmethod
+    def from_reference(entity_reference: EntityReference) -> LinklessEntitySpec:  # noqa: D102
+        return LinklessEntitySpec(element_name=entity_reference.element_name, entity_links=())

metricflow-semantics/metricflow_semantics/specs/group_by_metric_spec.py

@@ -0,0 +1,117 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any, Tuple
+
+from dbt_semantic_interfaces.dataclass_serialization import SerializableDataclass
+from dbt_semantic_interfaces.references import EntityReference
+from typing_extensions import override
+
+from metricflow_semantics.model.semantics.linkable_element import (
+    ElementPathKey,
+    GroupByMetricReference,
+    LinkableElementType,
+)
+from metricflow_semantics.naming.linkable_spec_name import StructuredLinkableSpecName
+from metricflow_semantics.specs.entity_spec import EntitySpec
+from metricflow_semantics.specs.instance_spec import InstanceSpecVisitor, LinkableInstanceSpec
+from metricflow_semantics.visitor import VisitorOutputT
+
+
+@dataclass(frozen=True)
+class GroupByMetricSpec(LinkableInstanceSpec, SerializableDataclass):
+    """Metric used in group by or where filter.
+
+    Args:
+        element_name: Name of the metric being joined.
+        entity_links: Sequence of entities joined to join the metric subquery to the outer query. Last entity is the one
+            joining the subquery to the outer query.
+        metric_subquery_entity_links: Sequence of entities used in the metric subquery to join the metric to the entity.
+    """
+
+    metric_subquery_entity_links: Tuple[EntityReference, ...]
+
+    def __post_init__(self) -> None:
+        """The inner query and outer query entity paths must end with the same entity (that's what they join on).
+
+        If no entity links, it's because we're already in the final joined node (no links left).
+        """
+        assert (
+            len(self.metric_subquery_entity_links) > 0
+        ), "GroupByMetricSpec must have at least one metric_subquery_entity_link."
+        if self.entity_links:
+            assert (
+                self.metric_subquery_entity_links[-1] == self.entity_links[-1]
+            ), "Inner and outer query must have the same last entity link in order to join on that link."
+
+    @property
+    def without_first_entity_link(self) -> GroupByMetricSpec:  # noqa: D102
+        assert len(self.entity_links) > 0, f"Spec does not have any entity links: {self}"
+        return GroupByMetricSpec(
+            element_name=self.element_name,
+            entity_links=self.entity_links[1:],
+            metric_subquery_entity_links=self.metric_subquery_entity_links,
+        )
+
+    @property
+    def without_entity_links(self) -> GroupByMetricSpec:  # noqa: D102
+        return GroupByMetricSpec(
+            element_name=self.element_name,
+            entity_links=(),
+            metric_subquery_entity_links=self.metric_subquery_entity_links,
+        )
+
+    @property
+    def last_entity_link(self) -> EntityReference:  # noqa: D102
+        assert len(self.entity_links) > 0, f"Spec does not have any entity links: {self}"
+        return self.entity_links[-1]
+
+    @property
+    def metric_subquery_entity_spec(self) -> EntitySpec:
+        """Spec for the entity that the metric will be grouped by in the metric subquery."""
+        assert (
+            len(self.metric_subquery_entity_links) > 0
+        ), "GroupByMetricSpec must have at least one metric_subquery_entity_link."
+        return EntitySpec(
+            element_name=self.metric_subquery_entity_links[-1].element_name,
+            entity_links=self.metric_subquery_entity_links[:-1],
+        )
+
+    @property
+    def qualified_name(self) -> str:
+        """Element name prefixed with entity links.
+
+        If same entity links are used in inner & outer query, use standard qualified name (country__bookings).
+        Else, specify both sets of entity links (listing__country__user__country__bookings).
+        """
+        if self.entity_links == self.metric_subquery_entity_links:
+            entity_links = self.entity_links
+        else:
+            entity_links = self.entity_links + self.metric_subquery_entity_links
+
+        return StructuredLinkableSpecName(
+            entity_link_names=tuple(entity_link.element_name for entity_link in entity_links),
+            element_name=self.element_name,
+        ).qualified_name
+
+    def __eq__(self, other: Any) -> bool:  # type: ignore[misc] # noqa: D105
+        if not isinstance(other, GroupByMetricSpec):
+            return False
+        return self.element_name == other.element_name and self.entity_links == other.entity_links
+
+    def __hash__(self) -> int:  # noqa: D105
+        return hash((self.element_name, self.entity_links, self.metric_subquery_entity_links))
+
+    @property
+    def reference(self) -> GroupByMetricReference:  # noqa: D102
+        return GroupByMetricReference(element_name=self.element_name)
+
+    def accept(self, visitor: InstanceSpecVisitor[VisitorOutputT]) -> VisitorOutputT:  # noqa: D102
+        return visitor.visit_group_by_metric_spec(self)
+
+    @property
+    @override
+    def element_path_key(self) -> ElementPathKey:
+        return ElementPathKey(
+            element_name=self.element_name, element_type=LinkableElementType.METRIC, entity_links=self.entity_links
+        )