refactor: route typed element methods through new filter helpers; add trio kwargs to get_elements_dataframe

get_elements_by_level and get_elements_filtered_by_wildcard are now thin
delegations to get_element_names with the appropriate kwargs. Behavior is
preserved (verified by regression tests against snapshots captured from
master before the refactor). Net effect: single source of truth for OData
$filter construction across the four element-listing methods.

get_elements_dataframe gains element_type / name_pattern / level kwargs.
When any of the trio is set while elements is None, the method resolves the
selection via get_element_names and feeds it into the existing MDX path.
The trio is authoritative and overrides skip_consolidations (documented in
the docstring).

Author: RobbyMeyers

TM1py/Services/ElementService.py

@@ -371,6 +371,9 @@ def get_elements_dataframe(
         allow_empty_alias: bool = True,
         attribute_suffix: bool = False,
         element_type_column: str = "Type",
+        element_type: Optional[Union[int, str, "Element.Types", Iterable]] = None,
+        name_pattern: Optional[str] = None,
+        level: Optional[int] = None,
         **kwargs,
     ) -> "pd.DataFrame":
         """
@@ -389,6 +392,16 @@ def get_elements_dataframe(
         :param allow_empty_alias: False if empty alias values should be substituted with element names instead
         :param attribute_suffix: True if attribute columns should have ':a', ':s' or ':n' suffix
         :param element_type_column: The column name in the df which specifies which element is which type.
+        :param element_type: Restrict to elements of the given type(s). Accepts an
+            ``Element.Types`` enum value, a string ('numeric'/'string'/'consolidated',
+            case-insensitive), an int (1/2/3), or an iterable of any of those.
+            Only applied when ``elements`` is None. When explicitly set, overrides
+            ``skip_consolidations``.
+        :param name_pattern: Restrict to elements whose name matches the glob pattern
+            (``*`` wildcard, case- and space-insensitive). Only applied when ``elements``
+            is None.
+        :param level: Restrict to elements at the given hierarchy level (0 = leaf).
+            Only applied when ``elements`` is None.
         :return: pandas DataFrame
         """
 
@@ -407,10 +420,39 @@ def get_elements_dataframe(
             unique_name = record[0][0]["UniqueName"]
             dimension_name, hierarchy_name, _ = dimension_hierarchy_element_tuple_from_unique_name(unique_name)
 
+        trio_filter_active = element_type is not None or name_pattern is not None or level is not None
         if elements is None or not any(elements):
-            elements = f"{{ [{dimension_name}].[{hierarchy_name}].Members }}"
-            if skip_consolidations:
-                elements = f"{{ Tm1FilterByLevel({elements}, 0) }}"
+            if trio_filter_active:
+                # Trio filter explicitly set. Resolve to a concrete element list via the
+                # filtered get_element_names path. The trio is authoritative and overrides
+                # skip_consolidations.
+                resolved = self.get_element_names(
+                    dimension_name=dimension_name,
+                    hierarchy_name=hierarchy_name,
+                    element_type=element_type,
+                    name_pattern=name_pattern,
+                    level=level,
+                )
+                if resolved:
+                    elements = (
+                        "{" + ",".join(f"[{dimension_name}].[{hierarchy_name}].[{member}]" for member in resolved) + "}"
+                    )
+                else:
+                    # Empty match. Filter the full Members set against an
+                    # unreachably high level so the MDX produces zero rows but the
+                    # downstream pipeline still emits the full column schema
+                    # (dimension name, attributes, levels, parents). A bare "{}"
+                    # axis would lose the dimension column and break the final
+                    # pd.merge on dimension_name.
+                    empty_set_level = 9999
+                    elements = (
+                        f"{{ Tm1FilterByLevel({{ [{dimension_name}].[{hierarchy_name}].Members }}, "
+                        f"{empty_set_level}) }}"
+                    )
+            else:
+                elements = f"{{ [{dimension_name}].[{hierarchy_name}].Members }}"
+                if skip_consolidations:
+                    elements = f"{{ Tm1FilterByLevel({elements}, 0) }}"
 
         if not isinstance(elements, str):
             if isinstance(elements, Iterable):
@@ -430,8 +472,13 @@ def get_elements_dataframe(
             )
         ]
 
+        # When the trio filter is active, the resolved element list is authoritative.
+        # Fetch the full type lookup so consolidated members survive the inner-join below.
+        element_types_skip_consolidations = False if trio_filter_active else skip_consolidations
         element_types = self.get_element_types(
-            dimension_name=dimension_name, hierarchy_name=hierarchy_name, skip_consolidations=skip_consolidations
+            dimension_name=dimension_name,
+            hierarchy_name=hierarchy_name,
+            skip_consolidations=element_types_skip_consolidations,
         )
 
         df = pd.DataFrame(
@@ -807,43 +854,33 @@ def get_all_leaf_element_identifiers(
         return self.get_element_identifiers(dimension_name, hierarchy_name, mdx_elements, **kwargs)
 
     def get_elements_by_level(self, dimension_name: str, hierarchy_name: str, level: int, **kwargs) -> List[str]:
-        """Get all element names by level in a hierarchy
+        """Get all element names by level in a hierarchy.
 
         :param dimension_name: Name of the dimension
         :param hierarchy_name: Name of the hierarchy
         :param level: Level to filter
         :return: List of element names
         """
-        url = format_url(
-            "/Dimensions('{}')/Hierarchies('{}')/Elements?$select=Name&$filter=Level eq {}",
-            dimension_name,
-            hierarchy_name,
-            str(level),
-        )
-        response = self._rest.GET(url, **kwargs)
-        return [e["Name"] for e in response.json()["value"]]
+        return self.get_element_names(dimension_name, hierarchy_name, level=level, **kwargs)
 
     def get_elements_filtered_by_wildcard(
         self, dimension_name: str, hierarchy_name: str, wildcard: str, level: int = None, **kwargs
     ) -> List[str]:
-        """Get all element names filtered by wildcard (CaseAndSpaceInsensitive) and level in a hierarchy
+        """Get all element names filtered by wildcard (case- and space-insensitive contains) and optional level.
 
         :param dimension_name: Name of the dimension
         :param hierarchy_name: Name of the hierarchy
-        :param wildcard: wildcard to filter
-        :param level: Level to filter
+        :param wildcard: substring to match (case- and space-insensitive contains)
+        :param level: Optional level to filter
         :return: List of element names
         """
-        filter_elements = format_url("contains(tolower(replace(Name,' ','')),tolower(replace('{}',' ', '')))", wildcard)
-        if level is not None:
-            filter_elements = filter_elements + f" and Level eq {level}"
-        url = format_url(
-            "/Dimensions('{}')/Hierarchies('{}')/Elements?$select=Name&$filter=" + filter_elements,
+        return self.get_element_names(
             dimension_name,
             hierarchy_name,
+            name_pattern=f"*{wildcard}*",
+            level=level,
+            **kwargs,
         )
-        response = self._rest.GET(url, **kwargs)
-        return [e["Name"] for e in response.json()["value"]]
 
     def get_all_element_identifiers(
         self, dimension_name: str, hierarchy_name: str, **kwargs

Tests/ElementService_test.py

@@ -1,5 +1,6 @@
 import configparser
 import copy
+import json
 import unittest
 from pathlib import Path
 
@@ -1668,6 +1669,10 @@ def setUp(self):
         h.add_edge("Total Regions", "Region North", 1)
         h.add_edge("Total Regions", "Region South", 1)
 
+        # Add a placeholder attribute so the }ElementAttributes_<dim> cube is
+        # created. get_elements_dataframe requires this cube to exist.
+        h.add_element_attribute("Description", "String")
+
         d.add_hierarchy(h)
         self.tm1.dimensions.update_or_create(d)
         self.addCleanup(self._cleanup_dimension)
@@ -1889,6 +1894,156 @@ def test_elements_quote_escape(self):
         self.assertEqual([e.name for e in elements], ["O'Brien"])
         self.assertEqual(elements[0].element_type, Element.Types.NUMERIC)
 
+    # ------------------------------------------------------------------
+    # Regression: verify behavior of typed methods is preserved after they
+    # are refactored to delegate to get_element_names. Snapshots in
+    # Tests/fixtures/element_filtering_snapshots/ were generated against
+    # master before the refactor.
+    # ------------------------------------------------------------------
+
+    SNAPSHOT_DIR = Path(__file__).parent / "fixtures" / "element_filtering_snapshots"
+
+    def _load_snapshot(self, name):
+        path = self.SNAPSHOT_DIR / name
+        if not path.exists():
+            self.fail(
+                f"Snapshot '{name}' not found at {self.SNAPSHOT_DIR}. "
+                f"Regenerate by re-running the snapshot generator from the plan's "
+                f"Phase 3 / Task 3.1."
+            )
+        with open(path) as f:
+            return json.load(f)
+
+    def test_regression_by_level_0(self):
+        actual = self.tm1.elements.get_elements_by_level(self.dimension_name, self.hierarchy_name, level=0)
+        expected = self._load_snapshot("by_level_0.json")
+        self.assertEqual(sorted(actual), expected)
+
+    def test_regression_by_level_1(self):
+        actual = self.tm1.elements.get_elements_by_level(self.dimension_name, self.hierarchy_name, level=1)
+        expected = self._load_snapshot("by_level_1.json")
+        self.assertEqual(sorted(actual), expected)
+
+    def test_regression_by_level_2(self):
+        actual = self.tm1.elements.get_elements_by_level(self.dimension_name, self.hierarchy_name, level=2)
+        expected = self._load_snapshot("by_level_2.json")
+        self.assertEqual(sorted(actual), expected)
+
+    def test_regression_wildcard_cases(self):
+        """Verify get_elements_filtered_by_wildcard preserves case+space-insensitive contains."""
+        for i in range(5):
+            snap = self._load_snapshot(f"wildcard_{i}.json")
+            actual = self.tm1.elements.get_elements_filtered_by_wildcard(
+                self.dimension_name,
+                self.hierarchy_name,
+                wildcard=snap["wildcard"],
+                level=snap["level"],
+            )
+            self.assertEqual(
+                sorted(actual),
+                snap["result"],
+                msg=(
+                    f"wildcard_{i}: wildcard={snap['wildcard']!r} level={snap['level']}, "
+                    f"got {sorted(actual)!r}, expected {snap['result']!r}"
+                ),
+            )
+
+    # ------------------------------------------------------------------
+    # get_elements_dataframe with trio kwargs
+    # ------------------------------------------------------------------
+
+    @skip_if_no_pandas
+    def test_dataframe_element_type_numeric(self):
+        df = self.tm1.elements.get_elements_dataframe(
+            self.dimension_name,
+            self.hierarchy_name,
+            element_type="numeric",
+            skip_consolidations=False,
+        )
+        names = set(df[self.dimension_name].tolist())
+        self.assertEqual(names, {"Numeric A", "Numeric B", "Numeric C", "O'Brien"})
+
+    @skip_if_no_pandas
+    def test_dataframe_pattern(self):
+        df = self.tm1.elements.get_elements_dataframe(
+            self.dimension_name,
+            self.hierarchy_name,
+            name_pattern="Region*",
+        )
+        names = set(df[self.dimension_name].tolist())
+        self.assertEqual(names, {"Region North", "Region South"})
+
+    @skip_if_no_pandas
+    def test_dataframe_level(self):
+        df = self.tm1.elements.get_elements_dataframe(
+            self.dimension_name,
+            self.hierarchy_name,
+            level=0,
+            skip_consolidations=False,
+        )
+        names = set(df[self.dimension_name].tolist())
+        self.assertEqual(
+            names,
+            {"Numeric A", "Numeric B", "Numeric C", "O'Brien", "String A", "String B"},
+        )
+
+    @skip_if_no_pandas
+    def test_dataframe_trio_composed(self):
+        df = self.tm1.elements.get_elements_dataframe(
+            self.dimension_name,
+            self.hierarchy_name,
+            element_type="numeric",
+            name_pattern="*A*",
+            level=0,
+        )
+        names = set(df[self.dimension_name].tolist())
+        self.assertEqual(names, {"Numeric A"})
+
+    @skip_if_no_pandas
+    def test_dataframe_element_type_overrides_skip_consolidations(self):
+        """When element_type is explicitly set, skip_consolidations is ignored
+        (documented in docstring)."""
+        df = self.tm1.elements.get_elements_dataframe(
+            self.dimension_name,
+            self.hierarchy_name,
+            element_type=["numeric", "consolidated"],
+            skip_consolidations=True,  # would normally drop consolidations
+        )
+        names = set(df[self.dimension_name].tolist())
+        # Consolidations should be present despite skip_consolidations=True
+        self.assertIn("Region North", names)
+        self.assertIn("Region South", names)
+        self.assertIn("Total Regions", names)
+
+    @skip_if_no_pandas
+    def test_dataframe_regression_no_filter(self):
+        """Without trio kwargs, get_elements_dataframe matches the snapshot from master."""
+        import pandas as pd
+
+        snapshot = pd.read_csv(self.SNAPSHOT_DIR / "dataframe_default.csv")
+        df = self.tm1.elements.get_elements_dataframe(self.dimension_name, self.hierarchy_name)
+        # Snapshot's first column is the snapshot's dimension name; the test's
+        # df uses a different dimension name. Compare row sets on element name + type.
+        snap_first = snapshot.columns[0]
+        df_first = df.columns[0]
+        snap_rows = sorted(zip(snapshot[snap_first].tolist(), snapshot["Type"].tolist()))
+        df_rows = sorted(zip(df[df_first].tolist(), df["Type"].tolist()))
+        self.assertEqual(snap_rows, df_rows)
+
+    @skip_if_no_pandas
+    def test_dataframe_trio_empty_match_preserves_schema(self):
+        """When the trio filter matches zero elements, the returned DataFrame must
+        still carry the full column schema (attributes, levels, parents) so callers
+        relying on df['<attr>'] don't see KeyError."""
+        df_full = self.tm1.elements.get_elements_dataframe(self.dimension_name, self.hierarchy_name)
+        df_empty = self.tm1.elements.get_elements_dataframe(
+            self.dimension_name,
+            self.hierarchy_name,
+            name_pattern="NonExistentNameThatMatchesNothing*",
+        )
+        self.assertEqual(list(df_full.columns), list(df_empty.columns))
+        self.assertEqual(len(df_empty), 0)
+
 
 if __name__ == "__main__":
     unittest.main()

Tests/fixtures/element_filtering_snapshots/by_level_0.json

@@ -0,0 +1,8 @@
+[
+  "Numeric A",
+  "Numeric B",
+  "Numeric C",
+  "O'Brien",
+  "String A",
+  "String B"
+]

Tests/fixtures/element_filtering_snapshots/by_level_1.json

@@ -0,0 +1,4 @@
+[
+  "Region North",
+  "Region South"
+]

Tests/fixtures/element_filtering_snapshots/by_level_2.json

@@ -0,0 +1,3 @@
+[
+  "Total Regions"
+]

Tests/fixtures/element_filtering_snapshots/dataframe_default.csv

@@ -0,0 +1,7 @@
+TM1py_snapshot_fixture,Type,Description,level001_Weight,level000_Weight,level001,level000
+String A,String,,0.000000,0.000000,,
+String B,String,,0.000000,0.000000,,
+Numeric A,Numeric,,1.000000,1.000000,Region North,Total Regions
+Numeric B,Numeric,,1.000000,1.000000,Region North,Total Regions
+Numeric C,Numeric,,1.000000,1.000000,Region South,Total Regions
+O'Brien,Numeric,,1.000000,1.000000,Region South,Total Regions

Tests/fixtures/element_filtering_snapshots/wildcard_0.json

@@ -0,0 +1,9 @@
+{
+  "wildcard": "region",
+  "level": null,
+  "result": [
+    "Region North",
+    "Region South",
+    "Total Regions"
+  ]
+}

Tests/fixtures/element_filtering_snapshots/wildcard_1.json

@@ -0,0 +1,9 @@
+{
+  "wildcard": "REGION",
+  "level": null,
+  "result": [
+    "Region North",
+    "Region South",
+    "Total Regions"
+  ]
+}

Tests/fixtures/element_filtering_snapshots/wildcard_2.json

@@ -0,0 +1,7 @@
+{
+  "wildcard": "Region North",
+  "level": null,
+  "result": [
+    "Region North"
+  ]
+}

Tests/fixtures/element_filtering_snapshots/wildcard_3.json

@@ -0,0 +1,9 @@
+{
+  "wildcard": "numeric",
+  "level": 0,
+  "result": [
+    "Numeric A",
+    "Numeric B",
+    "Numeric C"
+  ]
+}