Merge branch 'feature/PI-767-update_cpm_table_design' into release/2025-02-13

Author: megan-bower4

infrastructure/swagger/05_paths.yaml

@@ -315,7 +315,7 @@ paths:
       description: |
         - Delete a product using a product team ID and product ID.
       tags:
-        - Core Product ID Endpointsurce (DELETE)
+        - Core Product ID Endpoints
       parameters:
         - $ref: "#/components/parameters/ProductTeamId"
         - $ref: "#/components/parameters/ProductId"

src/api/readCpmProduct/tests/test_index.py

@@ -13,7 +13,7 @@
 
 TABLE_NAME = "hiya"
 ODS_CODE = "F5H1R"
-PRODUCT_TEAM_ID = "F5H1R.641be376-3954-4339-822c-54071c9ff1a0"
+PRODUCT_TEAM_ID = "641be376-3954-4339-822c-54071c9ff1a0"
 PRODUCT_TEAM_NAME = "product-team-name"
 PRODUCT_ID = "P.XXX-YYY"
 PRODUCT_NAME = "cpm-product-name"
@@ -64,7 +64,6 @@ def test_index(version):
                 },
             }
         )
-
     response_body = json_loads(result["body"])
 
     # Assertions for fields that must exactly match

src/api/readProductTeam/tests/test_index.py

@@ -23,7 +23,7 @@
 )
 def test_index(version):
     org = Root.create_ods_organisation(ods_code=CPM_PRODUCT_TEAM_NO_ID["ods_code"])
-    product_team = org.create_product_team_epr(
+    product_team = org.create_product_team(
         name=CPM_PRODUCT_TEAM_NO_ID["name"], keys=CPM_PRODUCT_TEAM_NO_ID["keys"]
     )
 
@@ -135,7 +135,7 @@ def test_index_no_such_product_team(version, product_id):
 )
 def test_index_by_alias(version):
     org = Root.create_ods_organisation(ods_code=CPM_PRODUCT_TEAM_NO_ID["ods_code"])
-    product_team = org.create_product_team_epr(
+    product_team = org.create_product_team(
         name=CPM_PRODUCT_TEAM_NO_ID["name"], keys=CPM_PRODUCT_TEAM_NO_ID["keys"]
     )
 

src/layers/domain/core/enum.py

@@ -12,3 +12,9 @@ class Environment(StrEnum):
     REF = auto()
     INT = auto()
     PROD = auto()
+
+
+class EntityType(StrEnum):
+    PRODUCT_TEAM = auto()
+    PRODUCT_TEAM_ALIAS = auto()
+    PRODUCT = auto()

src/layers/domain/repository/cpm_product_repository/tests/v1/test_cpm_product_repository_keys_v1.py

@@ -1,11 +1,7 @@
 import pytest
 from domain.core.cpm_product import CpmProduct
 from domain.core.product_key import ProductKey, ProductKeyType
-from domain.core.root import Root
 from domain.repository.cpm_product_repository import CpmProductRepository
-from domain.repository.errors import AlreadyExistsError
-
-from test_helpers.sample_data import CPM_PRODUCT_TEAM_NO_ID
 
 PARTY_KEY = "ABC-123456"
 
@@ -17,32 +13,7 @@ def test__product_repository__add_key(
     party_key = ProductKey(key_type=ProductKeyType.PARTY_KEY, key_value=PARTY_KEY)
     product.add_key(**party_key.dict())
     repository.write(product)
-
     product_by_id = repository.read(
         product_team_id=product.product_team_id, id=product.id
     )
     assert product_by_id.keys == [party_key]
-
-
-@pytest.mark.integration
-def test__product_repository__cannot_add_duplicate_key(
-    product: CpmProduct, repository: CpmProductRepository
-):
-    """This test guards against Party Key clashes"""
-
-    party_key = ProductKey(key_type=ProductKeyType.PARTY_KEY, key_value=PARTY_KEY)
-    product.add_key(**party_key.dict())
-    repository.write(product)
-
-    # Create a second unrelated product
-    org = Root.create_ods_organisation(ods_code=CPM_PRODUCT_TEAM_NO_ID["ods_code"])
-    second_product_team = org.create_product_team(
-        name=CPM_PRODUCT_TEAM_NO_ID["name"], keys=CPM_PRODUCT_TEAM_NO_ID["keys"]
-    )
-    second_product = second_product_team.create_cpm_product(
-        name="another-cpm-product-name"
-    )
-    second_product.add_key(**party_key.dict())
-
-    with pytest.raises(AlreadyExistsError):
-        repository.write(second_product)

src/layers/domain/repository/cpm_product_repository/v1.py

@@ -5,6 +5,7 @@
     CpmProductDeletedEvent,
     CpmProductKeyAddedEvent,
 )
+from domain.core.enum import EntityType
 from domain.core.product_key import ProductKey
 from domain.repository.cpm_repository import Repository
 from domain.repository.keys import TableKey
@@ -32,27 +33,27 @@ def handle_CpmProductCreatedEvent(self, event: CpmProductCreatedEvent):
             parent_key_parts=(event.product_team_id,),
             data=asdict(event),
             root=True,
+            row_type=EntityType.PRODUCT,
         )
 
     def handle_CpmProductKeyAddedEvent(self, event: CpmProductKeyAddedEvent):
         # Create a copy of the Product indexed against the new key
         new_key = ProductKey(**event.new_key)
-        create_transaction = self.create_index(
-            id=new_key.key_value,
-            parent_key_parts=(event.product_team_id,),
-            data=asdict(event),
-            root=False,
-        )
-
         # Update the value of "keys" on all other copies of this Device
         product_keys = {ProductKey(**key) for key in event.keys}
         product_keys_before_update = product_keys - {new_key}
         update_transactions = self.update_indexes(
+            pk=TableKey.PRODUCT_TEAM.key(event.product_team_id),
             id=event.id,
             keys=product_keys_before_update,
             data={"keys": event.keys, "updated_on": event.updated_on},
         )
-        return [create_transaction] + update_transactions
+        return update_transactions
 
     def handle_CpmProductDeletedEvent(self, event: CpmProductDeletedEvent):
-        return self.update_indexes(id=event.id, keys=event.keys, data=asdict(event))
+        return self.update_indexes(
+            pk=TableKey.PRODUCT_TEAM.key(event.product_team_id),
+            id=event.id,
+            keys=event.keys,
+            data=asdict(event),
+        )

src/layers/domain/repository/cpm_repository/tests/model_v1.py

@@ -72,6 +72,8 @@ def handle_MyEventAdd(self, event: MyEventAdd):
                     sk=MyTableKey.FOO.key(event.field),
                     pk_read_1=MyTableKey.FOO.key(event.field),
                     sk_read_1=MyTableKey.FOO.key(event.field),
+                    pk_read_2=MyTableKey.FOO.key(event.field),
+                    sk_read_2=MyTableKey.FOO.key(event.field),
                     **asdict(event)
                 ),
                 ConditionExpression=ConditionExpression.MUST_NOT_EXIST,
@@ -87,6 +89,8 @@ def handle_MyOtherEventAdd(self, event: MyOtherEventAdd):
                     sk=MyTableKey.BAR.key(event.field),
                     pk_read_1=MyTableKey.BAR.key(event.field),
                     sk_read_1=MyTableKey.BAR.key(event.field),
+                    pk_read_2=MyTableKey.BAR.key(event.field),
+                    sk_read_2=MyTableKey.BAR.key(event.field),
                     **asdict(event)
                 ),
             )

src/layers/domain/repository/cpm_repository/v1.py

@@ -3,6 +3,7 @@
 from typing import TYPE_CHECKING, Generator, Iterable
 
 from domain.core.aggregate_root import AggregateRoot
+from domain.core.enum import EntityType
 from domain.repository.errors import ItemNotFound
 from domain.repository.keys import KEY_SEPARATOR, TableKey
 from domain.repository.marshall import marshall, unmarshall
@@ -106,6 +107,7 @@ def create_index(
         parent_key_parts: tuple[str],
         data: dict,
         root: bool,
+        row_type: str,
         table_key: TableKey = None,
         parent_table_keys: tuple[TableKey] = None,
     ) -> TransactItem:
@@ -119,30 +121,37 @@ def create_index(
                 f"Expected provide {len(parent_table_keys)} parent key parts, got {len(parent_key_parts)}"
             )
 
-        write_key = table_key.key(id)
-        read_key = KEY_SEPARATOR.join(
+        sort_key = table_key.key(id)
+        partition_key = KEY_SEPARATOR.join(
             table_key.key(_id)
             for table_key, _id in zip(parent_table_keys, parent_key_parts)
         )
 
+        item_data = {
+            "pk": partition_key,
+            "sk": sort_key,
+            "pk_read_1": sort_key,
+            "sk_read_1": sort_key,
+            "root": root,
+            "row_type": row_type,
+            **data,
+        }
+
+        if row_type != EntityType.PRODUCT_TEAM_ALIAS:
+            item_data["pk_read_2"] = TableKey.ORG_CODE.key(data["ods_code"])
+            item_data["sk_read_2"] = sort_key
+
         return TransactItem(
             Put=TransactionStatement(
                 TableName=self.table_name,
-                Item=marshall(
-                    pk=write_key,
-                    sk=write_key,
-                    pk_read_1=read_key,
-                    sk_read_1=write_key,
-                    root=root,
-                    **data,
-                ),
+                Item=marshall(**item_data),
                 ConditionExpression=ConditionExpression.MUST_NOT_EXIST,
             )
         )
 
-    def update_indexes(self, id: str, keys: list[str], data: dict):
+    def update_indexes(self, pk: str, id: str, keys: list[str], data: dict):
         primary_keys = [
-            marshall(pk=pk, sk=pk) for pk in map(self.table_key.key, [id, *keys])
+            marshall(pk=pk, sk=sk) for sk in map(self.table_key.key, [id, *keys])
         ]
         return update_transactions(
             table_name=self.table_name, primary_keys=primary_keys, data=data
@@ -161,21 +170,18 @@ def delete_index(self, id: str):
     def _query(
         self, parent_ids: tuple[str], id: str = None, status: str = "all"
     ) -> list[dict]:
-        pk_read_1 = KEY_SEPARATOR.join(
+        pk = KEY_SEPARATOR.join(
             table_key.key(_id)
             for table_key, _id in zip(self.parent_table_keys, parent_ids)
         )
-        sk_read_1 = self.table_key.key(id or "")
+        sk = self.table_key.key(id or "")
 
         sk_query_type = QueryType.BEGINS_WITH if id is None else QueryType.EQUALS
-        sk_condition = sk_query_type.format("sk_read_1", ":sk_read_1")
+        sk_condition = sk_query_type.format("sk", ":sk")
         args = {
             "TableName": self.table_name,
-            "IndexName": "idx_gsi_read_1",
-            "KeyConditionExpression": f"pk_read_1 = :pk_read_1 AND {sk_condition}",
-            "ExpressionAttributeValues": marshall(
-                **{":pk_read_1": pk_read_1, ":sk_read_1": sk_read_1}
-            ),
+            "KeyConditionExpression": f"pk = :pk AND {sk_condition}",
+            "ExpressionAttributeValues": marshall(**{":pk": pk, ":sk": sk}),
         }
         if status != "all":
             args["FilterExpression"] = "#status = :status"
@@ -199,5 +205,7 @@ def _read(self, parent_ids: tuple[str], id: str, status: str = "all") -> ModelTy
         try:
             (item,) = items
         except ValueError:
+            if id in parent_ids:
+                raise ItemNotFound(id, item_type=self.model)
             raise ItemNotFound(*filter(bool, parent_ids), id, item_type=self.model)
         return self.model(**item)

src/layers/domain/repository/keys/v1.py

@@ -26,6 +26,7 @@ def filter_and_group(
 
 class TableKey(TableKeyAction, StrEnum):
     PRODUCT_TEAM = "PT"
+    PRODUCT_TEAM_ALIAS = "PTA"
     CPM_SYSTEM_ID = "CSI"
     CPM_PRODUCT = "P"
     CPM_PRODUCT_STATUS = "PS"
@@ -34,6 +35,7 @@ class TableKey(TableKeyAction, StrEnum):
     DEVICE_TAG = "DT"
     DEVICE_STATUS = "DS"
     ENVIRONMENT = "E"
+    ORG_CODE = "ORG"
 
 
 def group_by_key(

src/layers/domain/repository/product_team_repository/v1.py

@@ -1,4 +1,5 @@
 from attr import asdict
+from domain.core.enum import EntityType
 from domain.core.product_team import ProductTeam, ProductTeamCreatedEvent
 from domain.core.product_team_key import ProductTeamKey
 from domain.repository.cpm_repository import Repository
@@ -16,23 +17,28 @@ def __init__(self, table_name: str, dynamodb_client):
         )
 
     def read(self, id: str) -> ProductTeam:
-        return super()._read(parent_ids=("",), id=id, status="active")
+        return super()._read(parent_ids=(id,), id=id, status="active")
 
     def search(self) -> list[ProductTeam]:
         return super()._search(parent_ids=("",))
 
     def handle_ProductTeamCreatedEvent(self, event: ProductTeamCreatedEvent):
         create_root_transaction = self.create_index(
-            id=event.id, parent_key_parts=("",), data=asdict(event), root=True
+            id=event.id,
+            parent_key_parts=(event.id,),
+            data=asdict(event),
+            root=True,
+            row_type=EntityType.PRODUCT_TEAM,
         )
 
         keys = {ProductTeamKey(**key) for key in event.keys}
         create_key_transactions = [
             self.create_index(
                 id=key.key_value,
-                parent_key_parts=("",),
+                parent_key_parts=(key.key_value,),
                 data=asdict(event),
                 root=False,
+                row_type=EntityType.PRODUCT_TEAM_ALIAS,
             )
             for key in keys
         ]