[feature/PI-503-read_device_reference_data] read device reference data

Author: jaklinger

src/api/createDeviceReferenceData/tests/test_index.py

@@ -91,16 +91,13 @@ def test_index() -> None:
         repo = DeviceReferenceDataRepository(
             table_name=TABLE_NAME, dynamodb_client=index.cache["DYNAMODB_CLIENT"]
         )
-        if hasattr(repo, "read"):
-            raise Exception(
-                "uncomment the following code now that you've implemented read"
-            )
-            created_device_reference_data = repo.read(
-                product_team_id=device_reference_data.product_team_id,
-                product_id=device_reference_data.product_id,
-                device_reference_data_id=device_reference_data.id,
-            )
-            assert created_device_reference_data == device_reference_data
+
+        created_device_reference_data = repo.read(
+            product_team_id=device_reference_data.product_team_id,
+            product_id=device_reference_data.product_id,
+            device_reference_data_id=device_reference_data.id,
+        )
+        assert created_device_reference_data == device_reference_data
 
 
 @pytest.mark.parametrize(

src/api/readDeviceReferenceData/index.py

@@ -0,0 +1,26 @@
+from api_utils.api_step_chain import execute_step_chain
+from event.aws.client import dynamodb_client
+from event.environment import BaseEnvironment
+from event.logging.logger import setup_logger
+
+from .src.v1.steps import steps as v1_steps
+
+
+class Environment(BaseEnvironment):
+    DYNAMODB_TABLE: str
+
+
+versioned_steps = {"1": v1_steps}
+cache = {
+    **Environment.build().dict(),
+    "DYNAMODB_CLIENT": dynamodb_client(),
+}
+
+
+def handler(event: dict, context=None):
+    setup_logger(service_name=__file__)
+    return execute_step_chain(
+        event=event,
+        cache=cache,
+        versioned_steps=versioned_steps,
+    )

src/api/readDeviceReferenceData/make/make.py

@@ -0,0 +1,4 @@
+from builder.lambda_build import build
+
+if __name__ == "__main__":
+    build(__file__)

src/api/readDeviceReferenceData/policies/dynamodb.json

@@ -0,0 +1 @@
+["dynamodb:Query"]

src/api/readDeviceReferenceData/policies/kms.json

@@ -0,0 +1 @@
+["kms:Decrypt"]

src/api/readDeviceReferenceData/src/v1/steps.py

@@ -0,0 +1,63 @@
+from aws_lambda_powertools.utilities.data_classes import APIGatewayProxyEvent
+from domain.core.cpm_product.v1 import CpmProduct
+from domain.core.device_reference_data.v1 import DeviceReferenceData
+from domain.core.product_team.v3 import ProductTeam
+from domain.repository.cpm_product_repository.v3 import CpmProductRepository
+from domain.repository.device_reference_data_repository.v1 import (
+    DeviceReferenceDataRepository,
+)
+from domain.repository.product_team_repository.v2 import ProductTeamRepository
+from domain.request_models.v1 import DeviceReferenceDataPathParams
+from domain.response.validation_errors import mark_validation_errors_as_inbound
+from event.step_chain import StepChain
+
+
+@mark_validation_errors_as_inbound
+def parse_path_params(data, cache) -> DeviceReferenceDataPathParams:
+    event = APIGatewayProxyEvent(data[StepChain.INIT])
+    return DeviceReferenceDataPathParams(**event.path_parameters)
+
+
+def read_product_team(data, cache) -> ProductTeam:
+    path_params: DeviceReferenceDataPathParams = data[parse_path_params]
+    product_team_repo = ProductTeamRepository(
+        table_name=cache["DYNAMODB_TABLE"], dynamodb_client=cache["DYNAMODB_CLIENT"]
+    )
+    return product_team_repo.read(id=path_params.product_team_id)
+
+
+def read_product(data, cache) -> CpmProduct:
+    path_params: DeviceReferenceDataPathParams = data[parse_path_params]
+    product_repo = CpmProductRepository(
+        table_name=cache["DYNAMODB_TABLE"], dynamodb_client=cache["DYNAMODB_CLIENT"]
+    )
+    cpm_product = product_repo.read(
+        product_id=path_params.product_id, product_team_id=path_params.product_team_id
+    )
+    return cpm_product
+
+
+def read_device_reference_data(data, cache) -> DeviceReferenceData:
+    path_params: DeviceReferenceDataPathParams = data[parse_path_params]
+    product_repo = DeviceReferenceDataRepository(
+        table_name=cache["DYNAMODB_TABLE"], dynamodb_client=cache["DYNAMODB_CLIENT"]
+    )
+    return product_repo.read(
+        product_id=path_params.product_id,
+        product_team_id=path_params.product_team_id,
+        device_reference_data_id=path_params.device_reference_data_id,
+    )
+
+
+def device_reference_data_to_dict(data, cache) -> dict:
+    device_reference_data: DeviceReferenceData = data[read_device_reference_data]
+    return device_reference_data.state()
+
+
+steps = [
+    parse_path_params,
+    read_product_team,
+    read_product,
+    read_device_reference_data,
+    device_reference_data_to_dict,
+]