Add alternate key management (create, get, delete) for upsert support

tpellissier · claude · tpellissier · commit 179943976a64 · 2026-02-26T13:30:57.000-08:00
Adds client.tables.create_alternate_key(), get_alternate_keys(), and
delete_alternate_key() with AlternateKeyInfo typed return model.
Enables programmatic setup of alternate keys required for upsert operations.

Co-Authored-By: Claude Opus 4.6 (1M context) &lt;noreply@anthropic.com&gt;
diff --git a/src/PowerPlatform/Dataverse/data/_odata.py b/src/PowerPlatform/Dataverse/data/_odata.py
@@ -1486,6 +1486,102 @@ def _delete_table(self, table_schema_name: str) -> None:
         url = f"{self.api}/EntityDefinitions({metadata_id})"
         r = self._request("delete", url)
 
+    # ------------------- Alternate key metadata helpers -------------------
+
+    def _create_alternate_key(self, table_schema_name: str, key_name: str, columns: List[str]) -> Dict[str, Any]:
+        """Create an alternate key on a table.
+
+        Issues ``POST EntityDefinitions(LogicalName='{logical_name}')/Keys``
+        with payload ``{"SchemaName": key_name, "KeyAttributes": columns}``.
+
+        :param table_schema_name: Schema name of the table.
+        :type table_schema_name: ``str``
+        :param key_name: Schema name for the new alternate key.
+        :type key_name: ``str``
+        :param columns: List of column logical names that compose the key.
+        :type columns: ``list[str]``
+
+        :return: Dictionary with ``metadata_id``, ``schema_name``, and ``key_attributes``.
+        :rtype: ``dict[str, Any]``
+
+        :raises MetadataError: If the table does not exist.
+        :raises HttpError: If the Web API request fails.
+        """
+        ent = self._get_entity_by_table_schema_name(table_schema_name)
+        if not ent or not ent.get("MetadataId"):
+            raise MetadataError(
+                f"Table '{table_schema_name}' not found.",
+                subcode=METADATA_TABLE_NOT_FOUND,
+            )
+
+        logical_name = ent.get("LogicalName", table_schema_name.lower())
+        url = f"{self.api}/EntityDefinitions(LogicalName='{logical_name}')/Keys"
+        payload = {
+            "SchemaName": key_name,
+            "KeyAttributes": columns,
+        }
+        r = self._request("post", url, json=payload)
+        metadata_id = self._extract_id_from_header(r.headers.get("OData-EntityId"))
+
+        return {
+            "metadata_id": metadata_id,
+            "schema_name": key_name,
+            "key_attributes": columns,
+        }
+
+    def _get_alternate_keys(self, table_schema_name: str) -> List[Dict[str, Any]]:
+        """List all alternate keys on a table.
+
+        Issues ``GET EntityDefinitions(LogicalName='{logical_name}')/Keys``.
+
+        :param table_schema_name: Schema name of the table.
+        :type table_schema_name: ``str``
+
+        :return: List of raw ``EntityKeyMetadata`` dictionaries.
+        :rtype: ``list[dict[str, Any]]``
+
+        :raises MetadataError: If the table does not exist.
+        :raises HttpError: If the Web API request fails.
+        """
+        ent = self._get_entity_by_table_schema_name(table_schema_name)
+        if not ent or not ent.get("MetadataId"):
+            raise MetadataError(
+                f"Table '{table_schema_name}' not found.",
+                subcode=METADATA_TABLE_NOT_FOUND,
+            )
+
+        logical_name = ent.get("LogicalName", table_schema_name.lower())
+        url = f"{self.api}/EntityDefinitions(LogicalName='{logical_name}')/Keys"
+        r = self._request("get", url)
+        return r.json().get("value", [])
+
+    def _delete_alternate_key(self, table_schema_name: str, key_id: str) -> None:
+        """Delete an alternate key by metadata ID.
+
+        Issues ``DELETE EntityDefinitions(LogicalName='{logical_name}')/Keys({key_id})``.
+
+        :param table_schema_name: Schema name of the table.
+        :type table_schema_name: ``str``
+        :param key_id: Metadata GUID of the alternate key.
+        :type key_id: ``str``
+
+        :return: ``None``
+        :rtype: ``None``
+
+        :raises MetadataError: If the table does not exist.
+        :raises HttpError: If the Web API request fails.
+        """
+        ent = self._get_entity_by_table_schema_name(table_schema_name)
+        if not ent or not ent.get("MetadataId"):
+            raise MetadataError(
+                f"Table '{table_schema_name}' not found.",
+                subcode=METADATA_TABLE_NOT_FOUND,
+            )
+
+        logical_name = ent.get("LogicalName", table_schema_name.lower())
+        url = f"{self.api}/EntityDefinitions(LogicalName='{logical_name}')/Keys({key_id})"
+        self._request("delete", url)
+
     def _create_table(
         self,
         table_schema_name: str,
diff --git a/src/PowerPlatform/Dataverse/models/table_info.py b/src/PowerPlatform/Dataverse/models/table_info.py
@@ -0,0 +1,47 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT license.
+
+"""Table metadata models for Dataverse."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any, Dict, List
+
+
+@dataclass
+class AlternateKeyInfo:
+    """Alternate key metadata for a Dataverse table.
+
+    :param metadata_id: Key metadata GUID.
+    :type metadata_id: :class:`str`
+    :param schema_name: Key schema name.
+    :type schema_name: :class:`str`
+    :param key_attributes: List of column logical names that compose the key.
+    :type key_attributes: :class:`list` of :class:`str`
+    :param status: Index creation status (``"Active"``, ``"Pending"``, ``"InProgress"``, ``"Failed"``).
+    :type status: :class:`str`
+    """
+
+    metadata_id: str = ""
+    schema_name: str = ""
+    key_attributes: List[str] = field(default_factory=list)
+    status: str = ""
+
+    @classmethod
+    def from_api_response(cls, response_data: Dict[str, Any]) -> AlternateKeyInfo:
+        """Create from raw EntityKeyMetadata API response.
+
+        :param response_data: Raw key metadata dictionary from the Web API.
+        :type response_data: :class:`dict`
+        :rtype: :class:`AlternateKeyInfo`
+        """
+        return cls(
+            metadata_id=response_data.get("MetadataId", ""),
+            schema_name=response_data.get("SchemaName", ""),
+            key_attributes=response_data.get("KeyAttributes", []),
+            status=response_data.get("EntityKeyIndexStatus", ""),
+        )
+
+
+__all__ = ["AlternateKeyInfo"]
diff --git a/src/PowerPlatform/Dataverse/operations/tables.py b/src/PowerPlatform/Dataverse/operations/tables.py
@@ -14,6 +14,7 @@
     CascadeConfiguration,
     RelationshipInfo,
 )
+from ..models.table_info import AlternateKeyInfo
 from ..models.labels import Label, LocalizedLabel
 from ..common.constants import CASCADE_BEHAVIOR_REMOVE_LINK
 
@@ -578,3 +579,111 @@ def create_lookup_field(
         )
 
         return self.create_one_to_many_relationship(lookup, relationship, solution=solution)
+
+    # ------------------------------------------------- create_alternate_key
+
+    def create_alternate_key(
+        self,
+        table: str,
+        key_name: str,
+        columns: List[str],
+    ) -> AlternateKeyInfo:
+        """Create an alternate key on a table.
+
+        Alternate keys allow upsert operations to identify records by one or
+        more columns instead of the primary GUID. After creation the key is
+        queued for index building; its :attr:`~AlternateKeyInfo.status` will
+        transition from ``"Pending"`` to ``"Active"`` once the index is ready.
+
+        :param table: Schema name of the table (e.g. ``"new_Product"``).
+        :type table: :class:`str`
+        :param key_name: Schema name for the new alternate key
+            (e.g. ``"new_product_code_key"``).
+        :type key_name: :class:`str`
+        :param columns: List of column logical names that compose the key
+            (e.g. ``["new_productcode"]``).
+        :type columns: :class:`list` of :class:`str`
+
+        :return: Metadata for the newly created alternate key.
+        :rtype: :class:`~PowerPlatform.Dataverse.models.table_info.AlternateKeyInfo`
+
+        :raises ~PowerPlatform.Dataverse.core.errors.MetadataError:
+            If the table does not exist.
+        :raises ~PowerPlatform.Dataverse.core.errors.HttpError:
+            If the Web API request fails.
+
+        Example:
+            Create a single-column alternate key for upsert::
+
+                key = client.tables.create_alternate_key(
+                    "new_Product",
+                    "new_product_code_key",
+                    ["new_productcode"],
+                )
+                print(f"Key ID: {key.metadata_id}")
+                print(f"Columns: {key.key_attributes}")
+        """
+        with self._client._scoped_odata() as od:
+            raw = od._create_alternate_key(table, key_name, columns)
+            return AlternateKeyInfo(
+                metadata_id=raw["metadata_id"],
+                schema_name=raw["schema_name"],
+                key_attributes=raw["key_attributes"],
+            )
+
+    # --------------------------------------------------- get_alternate_keys
+
+    def get_alternate_keys(self, table: str) -> List[AlternateKeyInfo]:
+        """List all alternate keys defined on a table.
+
+        :param table: Schema name of the table (e.g. ``"new_Product"``).
+        :type table: :class:`str`
+
+        :return: List of alternate key metadata objects. May be empty if no
+            alternate keys are defined.
+        :rtype: :class:`list` of :class:`~PowerPlatform.Dataverse.models.table_info.AlternateKeyInfo`
+
+        :raises ~PowerPlatform.Dataverse.core.errors.MetadataError:
+            If the table does not exist.
+        :raises ~PowerPlatform.Dataverse.core.errors.HttpError:
+            If the Web API request fails.
+
+        Example:
+            List alternate keys and print their status::
+
+                keys = client.tables.get_alternate_keys("new_Product")
+                for key in keys:
+                    print(f"{key.schema_name}: {key.status}")
+        """
+        with self._client._scoped_odata() as od:
+            raw_list = od._get_alternate_keys(table)
+            return [AlternateKeyInfo.from_api_response(item) for item in raw_list]
+
+    # ------------------------------------------------ delete_alternate_key
+
+    def delete_alternate_key(self, table: str, key_id: str) -> None:
+        """Delete an alternate key by its metadata ID.
+
+        :param table: Schema name of the table (e.g. ``"new_Product"``).
+        :type table: :class:`str`
+        :param key_id: Metadata GUID of the alternate key to delete.
+        :type key_id: :class:`str`
+
+        :raises ~PowerPlatform.Dataverse.core.errors.MetadataError:
+            If the table does not exist.
+        :raises ~PowerPlatform.Dataverse.core.errors.HttpError:
+            If the Web API request fails.
+
+        .. warning::
+            Deleting an alternate key that is in use by upsert operations will
+            cause those operations to fail. This operation is irreversible.
+
+        Example::
+
+            client.tables.delete_alternate_key(
+                "new_Product",
+                "12345678-1234-1234-1234-123456789abc",
+            )
+        """
+        with self._client._scoped_odata() as od:
+            od._delete_alternate_key(table, key_id)
diff --git a/tests/unit/models/test_alternate_key.py b/tests/unit/models/test_alternate_key.py
@@ -0,0 +1,80 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT license.
+
+import unittest
+
+from PowerPlatform.Dataverse.models.table_info import AlternateKeyInfo
+
+
+class TestAlternateKeyInfoDefaults(unittest.TestCase):
+    """Tests for AlternateKeyInfo default values."""
+
+    def test_default_values(self):
+        """All fields should default to empty string or empty list."""
+        info = AlternateKeyInfo()
+        self.assertEqual(info.metadata_id, "")
+        self.assertEqual(info.schema_name, "")
+        self.assertEqual(info.key_attributes, [])
+        self.assertEqual(info.status, "")
+
+    def test_independent_default_lists(self):
+        """Each instance should have its own key_attributes list (no shared mutable default)."""
+        a = AlternateKeyInfo()
+        b = AlternateKeyInfo()
+        a.key_attributes.append("col1")
+        self.assertEqual(b.key_attributes, [])
+
+
+class TestAlternateKeyInfoFromApiResponse(unittest.TestCase):
+    """Tests for AlternateKeyInfo.from_api_response factory."""
+
+    def test_full_response(self):
+        """from_api_response should map all PascalCase API fields."""
+        raw = {
+            "MetadataId": "key-guid-1",
+            "SchemaName": "new_product_code_key",
+            "KeyAttributes": ["new_productcode"],
+            "EntityKeyIndexStatus": "Active",
+        }
+        info = AlternateKeyInfo.from_api_response(raw)
+        self.assertEqual(info.metadata_id, "key-guid-1")
+        self.assertEqual(info.schema_name, "new_product_code_key")
+        self.assertEqual(info.key_attributes, ["new_productcode"])
+        self.assertEqual(info.status, "Active")
+
+    def test_multi_column_key(self):
+        """from_api_response should handle multi-column keys."""
+        raw = {
+            "MetadataId": "key-guid-2",
+            "SchemaName": "new_composite_key",
+            "KeyAttributes": ["new_col1", "new_col2", "new_col3"],
+            "EntityKeyIndexStatus": "Pending",
+        }
+        info = AlternateKeyInfo.from_api_response(raw)
+        self.assertEqual(info.key_attributes, ["new_col1", "new_col2", "new_col3"])
+        self.assertEqual(info.status, "Pending")
+
+    def test_minimal_response(self):
+        """from_api_response should handle a response with missing optional fields."""
+        raw = {}
+        info = AlternateKeyInfo.from_api_response(raw)
+        self.assertEqual(info.metadata_id, "")
+        self.assertEqual(info.schema_name, "")
+        self.assertEqual(info.key_attributes, [])
+        self.assertEqual(info.status, "")
+
+    def test_partial_response(self):
+        """from_api_response should handle a response with only some fields."""
+        raw = {
+            "MetadataId": "key-guid-3",
+            "SchemaName": "new_partial_key",
+        }
+        info = AlternateKeyInfo.from_api_response(raw)
+        self.assertEqual(info.metadata_id, "key-guid-3")
+        self.assertEqual(info.schema_name, "new_partial_key")
+        self.assertEqual(info.key_attributes, [])
+        self.assertEqual(info.status, "")
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tests/unit/test_tables_operations.py b/tests/unit/test_tables_operations.py
