Merge remote-tracking branch 'origin/main' into feature/relationship-info-model

Author: tpellissier

examples/basic/functional_testing.py

@@ -65,6 +65,13 @@ def setup_authentication() -> DataverseClient:
         print("Testing connection...")
         tables = client.tables.list()
         print(f"[OK] Connection successful! Found {len(tables)} tables.")
+
+        # Test filtered + selected list
+        user_owned = client.tables.list(
+            filter="OwnershipType eq Microsoft.Dynamics.CRM.OwnershipTypes'UserOwned'",
+            select=["LogicalName", "SchemaName", "DisplayName"],
+        )
+        print(f"[OK] Found {len(user_owned)} user-owned tables (filter + select).")
         return client
 
     except Exception as e:

examples/basic/installation_example.py

@@ -265,6 +265,13 @@ def show_usage_examples():
 # List all tables
 tables = client.tables.list()
 print(f"Found {len(tables)} tables")
+
+# List with filter and select
+custom_tables = client.tables.list(
+    filter="IsCustomEntity eq true",
+    select=["LogicalName", "SchemaName", "DisplayName"],
+)
+print(f"Found {len(custom_tables)} custom tables")
 ```
 """)
 
@@ -304,9 +311,14 @@ def interactive_test():
 
         print("  Testing connection...")
         tables = client.tables.list()
-
         print(f"  [OK] Connection successful!")
         print(f"  Found {len(tables)} tables in environment")
+
+        custom_tables = client.tables.list(
+            filter="IsCustomEntity eq true",
+            select=["LogicalName", "SchemaName"],
+        )
+        print(f"  Found {len(custom_tables)} custom tables (filter + select)")
         print(f"  Connected to: {org_url}")
 
         print("\n  Your SDK is ready for use!")

src/PowerPlatform/Dataverse/data/_odata.py

@@ -1423,16 +1423,44 @@ def _get_table_info(self, table_schema_name: str) -> Optional[Dict[str, Any]]:
             "columns_created": [],
         }
 
-    def _list_tables(self) -> List[Dict[str, Any]]:
+    def _list_tables(
+        self,
+        filter: Optional[str] = None,
+        select: Optional[List[str]] = None,
+    ) -> List[Dict[str, Any]]:
         """List all non-private tables (``IsPrivate eq false``).
 
+        :param filter: Optional additional OData ``$filter`` expression that is
+            combined with the default ``IsPrivate eq false`` clause using
+            ``and``.  For example, ``"SchemaName eq 'Account'"`` becomes
+            ``"IsPrivate eq false and (SchemaName eq 'Account')"``.
+            When ``None`` (the default), only the ``IsPrivate eq false`` filter
+            is applied.
+        :type filter: ``str`` or ``None``
+        :param select: Optional list of property names to project via
+            ``$select``.  Values are passed as-is (PascalCase) because
+            ``EntityDefinitions`` uses PascalCase property names.
+            When ``None`` (the default) or an empty list, no ``$select`` is
+            applied and all properties are returned.  Passing a bare string
+            raises ``TypeError``.
+        :type select: ``list[str]`` or ``None``
+
         :return: Metadata entries for non-private tables (may be empty).
         :rtype: ``list[dict[str, Any]]``
 
         :raises HttpError: If the metadata request fails.
         """
         url = f"{self.api}/EntityDefinitions"
-        params = {"$filter": "IsPrivate eq false"}
+        base_filter = "IsPrivate eq false"
+        if filter:
+            combined_filter = f"{base_filter} and ({filter})"
+        else:
+            combined_filter = base_filter
+        params: Dict[str, str] = {"$filter": combined_filter}
+        if select is not None and isinstance(select, str):
+            raise TypeError("select must be a list of property names, not a bare string")
+        if select:
+            params["$select"] = ",".join(select)
         r = self._request("get", url, params=params)
         return r.json().get("value", [])
 

src/PowerPlatform/Dataverse/operations/tables.py

@@ -179,20 +179,56 @@ def get(self, table: str) -> Optional[Dict[str, Any]]:
 
     # ------------------------------------------------------------------- list
 
-    def list(self) -> List[Dict[str, Any]]:
+    def list(
+        self,
+        *,
+        filter: Optional[str] = None,
+        select: Optional[List[str]] = None,
+    ) -> List[Dict[str, Any]]:
         """List all non-private tables in the Dataverse environment.
 
+        By default returns every table where ``IsPrivate eq false``.  Supply
+        an optional OData ``$filter`` expression to further narrow the results.
+        The expression is combined with the default ``IsPrivate eq false``
+        clause using ``and``.
+
+        :param filter: Optional OData ``$filter`` expression to further narrow
+            the list of returned tables (e.g.
+            ``"SchemaName eq 'Account'"``).  Column names in filter
+            expressions must use the exact property names from the
+            ``EntityDefinitions`` metadata (typically PascalCase).
+        :type filter: :class:`str` or None
+        :param select: Optional list of property names to include in the
+            response (projected via the OData ``$select`` query option).
+            Property names must use the exact PascalCase names from the
+            ``EntityDefinitions`` metadata (e.g.
+            ``["LogicalName", "SchemaName", "DisplayName"]``).
+            When ``None`` (the default) or an empty list, all properties are
+            returned.
+        :type select: :class:`list` of :class:`str` or None
+
         :return: List of EntityDefinition metadata dictionaries.
         :rtype: :class:`list` of :class:`dict`
 
         Example::
 
+            # List all non-private tables
             tables = client.tables.list()
             for table in tables:
                 print(table["LogicalName"])
+
+            # List only tables whose schema name starts with "new_"
+            custom_tables = client.tables.list(
+                filter="startswith(SchemaName, 'new_')"
+            )
+
+            # List tables with only specific properties
+            tables = client.tables.list(
+                select=["LogicalName", "SchemaName", "EntitySetName"]
+            )
         """
         with self._client._scoped_odata() as od:
-            return od._list_tables()
+            return od._list_tables(filter=filter, select=select)
 
     # ------------------------------------------------------------- add_columns
 

tests/unit/data/test_odata_internal.py

@@ -125,6 +125,136 @@ def test_non_string_key_raises_type_error(self):
             self.od._build_alternate_key_str({1: "ACC-001"})
 
 
+class TestListTables(unittest.TestCase):
+    """Unit tests for _ODataClient._list_tables filter and select parameters."""
+
+    def setUp(self):
+        self.od = _make_odata_client()
+
+    def _setup_response(self, value):
+        """Configure _request to return a response with the given value list."""
+        mock_response = MagicMock()
+        mock_response.json.return_value = {"value": value}
+        self.od._request.return_value = mock_response
+
+    def test_no_filter_uses_default(self):
+        """_list_tables() without filter sends only IsPrivate eq false."""
+        self._setup_response([])
+        self.od._list_tables()
+
+        self.od._request.assert_called_once()
+        call_kwargs = self.od._request.call_args
+        params = call_kwargs.kwargs.get("params") or call_kwargs[1].get("params", {})
+        self.assertEqual(params["$filter"], "IsPrivate eq false")
+
+    def test_filter_combined_with_default(self):
+        """_list_tables(filter=...) combines user filter with IsPrivate eq false."""
+        self._setup_response([{"LogicalName": "account"}])
+        self.od._list_tables(filter="SchemaName eq 'Account'")
+
+        self.od._request.assert_called_once()
+        call_kwargs = self.od._request.call_args
+        params = call_kwargs.kwargs.get("params") or call_kwargs[1].get("params", {})
+        self.assertEqual(
+            params["$filter"],
+            "IsPrivate eq false and (SchemaName eq 'Account')",
+        )
+
+    def test_filter_none_same_as_no_filter(self):
+        """_list_tables(filter=None) is equivalent to _list_tables()."""
+        self._setup_response([])
+        self.od._list_tables(filter=None)
+
+        self.od._request.assert_called_once()
+        call_kwargs = self.od._request.call_args
+        params = call_kwargs.kwargs.get("params") or call_kwargs[1].get("params", {})
+        self.assertEqual(params["$filter"], "IsPrivate eq false")
+
+    def test_returns_value_list(self):
+        """_list_tables returns the 'value' array from the response."""
+        expected = [
+            {"LogicalName": "account"},
+            {"LogicalName": "contact"},
+        ]
+        self._setup_response(expected)
+        result = self.od._list_tables()
+        self.assertEqual(result, expected)
+
+    def test_select_adds_query_param(self):
+        """_list_tables(select=...) adds $select as comma-joined string."""
+        self._setup_response([])
+        self.od._list_tables(select=["LogicalName", "SchemaName", "DisplayName"])
+
+        self.od._request.assert_called_once()
+        call_kwargs = self.od._request.call_args
+        params = call_kwargs.kwargs.get("params") or call_kwargs[1].get("params", {})
+        self.assertEqual(params["$select"], "LogicalName,SchemaName,DisplayName")
+
+    def test_select_none_omits_query_param(self):
+        """_list_tables(select=None) does not add $select to params."""
+        self._setup_response([])
+        self.od._list_tables(select=None)
+
+        self.od._request.assert_called_once()
+        call_kwargs = self.od._request.call_args
+        params = call_kwargs.kwargs.get("params") or call_kwargs[1].get("params", {})
+        self.assertNotIn("$select", params)
+
+    def test_select_empty_list_omits_query_param(self):
+        """_list_tables(select=[]) does not add $select (empty list is falsy)."""
+        self._setup_response([])
+        self.od._list_tables(select=[])
+
+        self.od._request.assert_called_once()
+        call_kwargs = self.od._request.call_args
+        params = call_kwargs.kwargs.get("params") or call_kwargs[1].get("params", {})
+        self.assertNotIn("$select", params)
+
+    def test_select_preserves_case(self):
+        """_list_tables does not lowercase select values (PascalCase preserved)."""
+        self._setup_response([])
+        self.od._list_tables(select=["EntitySetName", "LogicalName"])
+
+        self.od._request.assert_called_once()
+        call_kwargs = self.od._request.call_args
+        params = call_kwargs.kwargs.get("params") or call_kwargs[1].get("params", {})
+        self.assertEqual(params["$select"], "EntitySetName,LogicalName")
+
+    def test_select_with_filter(self):
+        """_list_tables with both select and filter sends both params."""
+        self._setup_response([{"LogicalName": "account"}])
+        self.od._list_tables(
+            filter="SchemaName eq 'Account'",
+            select=["LogicalName", "SchemaName"],
+        )
+
+        self.od._request.assert_called_once()
+        call_kwargs = self.od._request.call_args
+        params = call_kwargs.kwargs.get("params") or call_kwargs[1].get("params", {})
+        self.assertEqual(
+            params["$filter"],
+            "IsPrivate eq false and (SchemaName eq 'Account')",
+        )
+        self.assertEqual(params["$select"], "LogicalName,SchemaName")
+
+    def test_select_single_property(self):
+        """_list_tables(select=[...]) with a single property works correctly."""
+        self._setup_response([])
+        self.od._list_tables(select=["LogicalName"])
+
+        self.od._request.assert_called_once()
+        call_kwargs = self.od._request.call_args
+        params = call_kwargs.kwargs.get("params") or call_kwargs[1].get("params", {})
+        self.assertEqual(params["$select"], "LogicalName")
+
+    def test_select_bare_string_raises_type_error(self):
+        """_list_tables(select='LogicalName') raises TypeError for bare str."""
+        self._setup_response([])
+        with self.assertRaises(TypeError) as ctx:
+            self.od._list_tables(select="LogicalName")
+        self.assertIn("list of property names", str(ctx.exception))
+
+
 class TestUpsert(unittest.TestCase):
     """Unit tests for _ODataClient._upsert."""
 

tests/unit/test_tables_operations.py

@@ -101,10 +101,80 @@ def test_list(self):
 
         result = self.client.tables.list()
 
-        self.client._odata._list_tables.assert_called_once()
+        self.client._odata._list_tables.assert_called_once_with(filter=None, select=None)
         self.assertIsInstance(result, list)
         self.assertEqual(result, expected_tables)
 
+    def test_list_with_filter(self):
+        """list(filter=...) should pass the filter expression to _list_tables."""
+        expected_tables = [
+            {"LogicalName": "account", "SchemaName": "Account"},
+        ]
+        self.client._odata._list_tables.return_value = expected_tables
+
+        result = self.client.tables.list(filter="SchemaName eq 'Account'")
+
+        self.client._odata._list_tables.assert_called_once_with(filter="SchemaName eq 'Account'", select=None)
+        self.assertIsInstance(result, list)
+        self.assertEqual(result, expected_tables)
+
+    def test_list_with_filter_none_explicit(self):
+        """list(filter=None) should behave identically to list() with no args."""
+        expected_tables = [
+            {"LogicalName": "account", "SchemaName": "Account"},
+        ]
+        self.client._odata._list_tables.return_value = expected_tables
+
+        result = self.client.tables.list(filter=None)
+
+        self.client._odata._list_tables.assert_called_once_with(filter=None, select=None)
+        self.assertEqual(result, expected_tables)
+
+    def test_list_with_select(self):
+        """list(select=...) should pass the select list to _list_tables."""
+        expected_tables = [
+            {"LogicalName": "account", "SchemaName": "Account"},
+        ]
+        self.client._odata._list_tables.return_value = expected_tables
+
+        result = self.client.tables.list(select=["LogicalName", "SchemaName", "EntitySetName"])
+
+        self.client._odata._list_tables.assert_called_once_with(
+            filter=None,
+            select=["LogicalName", "SchemaName", "EntitySetName"],
+        )
+        self.assertEqual(result, expected_tables)
+
+    def test_list_with_select_none_explicit(self):
+        """list(select=None) should behave identically to list() with no args."""
+        expected_tables = [
+            {"LogicalName": "account", "SchemaName": "Account"},
+        ]
+        self.client._odata._list_tables.return_value = expected_tables
+
+        result = self.client.tables.list(select=None)
+
+        self.client._odata._list_tables.assert_called_once_with(filter=None, select=None)
+        self.assertEqual(result, expected_tables)
+
+    def test_list_with_filter_and_select(self):
+        """list(filter=..., select=...) should pass both params to _list_tables."""
+        expected_tables = [
+            {"LogicalName": "account", "SchemaName": "Account"},
+        ]
+        self.client._odata._list_tables.return_value = expected_tables
+
+        result = self.client.tables.list(
+            filter="SchemaName eq 'Account'",
+            select=["LogicalName", "SchemaName"],
+        )
+
+        self.client._odata._list_tables.assert_called_once_with(
+            filter="SchemaName eq 'Account'",
+            select=["LogicalName", "SchemaName"],
+        )
+        self.assertEqual(result, expected_tables)
+
     # ------------------------------------------------------------ add_columns
 
     def test_add_columns(self):