feat(data-classes): Add connect contact flow event (#304)

aws_lambda_powertools/utilities/data_classes/__init__.py

@@ -1,6 +1,7 @@
 from .alb_event import ALBEvent
 from .api_gateway_proxy_event import APIGatewayProxyEvent, APIGatewayProxyEventV2
 from .cloud_watch_logs_event import CloudWatchLogsEvent
+from .connect_contact_flow_event import ConnectContactFlowEvent
 from .dynamo_db_stream_event import DynamoDBStreamEvent
 from .event_bridge_event import EventBridgeEvent
 from .kinesis_stream_event import KinesisStreamEvent
@@ -14,6 +15,7 @@
     "APIGatewayProxyEventV2",
     "ALBEvent",
     "CloudWatchLogsEvent",
+    "ConnectContactFlowEvent",
     "DynamoDBStreamEvent",
     "EventBridgeEvent",
     "KinesisStreamEvent",

aws_lambda_powertools/utilities/data_classes/connect_contact_flow_event.py

@@ -0,0 +1,166 @@
+from enum import Enum, auto
+from typing import Dict, Optional
+
+from aws_lambda_powertools.utilities.data_classes.common import DictWrapper
+
+
+class ConnectContactFlowChannel(Enum):
+    VOICE = auto()
+    CHAT = auto()
+
+
+class ConnectContactFlowEndpointType(Enum):
+    TELEPHONE_NUMBER = auto()
+
+
+class ConnectContactFlowInitiationMethod(Enum):
+    INBOUND = auto()
+    OUTBOUND = auto()
+    TRANSFER = auto()
+    CALLBACK = auto()
+    API = auto()
+
+
+class ConnectContactFlowEndpoint(DictWrapper):
+    @property
+    def address(self) -> str:
+        """The phone number."""
+        return self["Address"]
+
+    @property
+    def endpoint_type(self) -> ConnectContactFlowEndpointType:
+        """The enpoint type."""
+        return ConnectContactFlowEndpointType[self["Type"]]
+
+
+class ConnectContactFlowQueue(DictWrapper):
+    @property
+    def arn(self) -> str:
+        """The unique queue ARN."""
+        return self["ARN"]
+
+    @property
+    def name(self) -> str:
+        """The queue name."""
+        return self["Name"]
+
+
+class ConnectContactFlowMediaStreamAudio(DictWrapper):
+    @property
+    def start_fragment_number(self) -> Optional[str]:
+        """The number that identifies the Kinesis Video Streams fragment, in the stream used for Live media streaming,
+        in which the customer audio stream started.
+        """
+        return self["StartFragmentNumber"]
+
+    @property
+    def start_timestamp(self) -> Optional[str]:
+        """When the customer audio stream started."""
+        return self["StartTimestamp"]
+
+    @property
+    def stream_arn(self) -> Optional[str]:
+        """The ARN of the Kinesis Video stream used for Live media streaming that includes the customer data to
+        reference.
+        """
+        return self["StreamARN"]
+
+
+class ConnectContactFlowMediaStreamCustomer(DictWrapper):
+    @property
+    def audio(self) -> ConnectContactFlowMediaStreamAudio:
+        return ConnectContactFlowMediaStreamAudio(self["Audio"])
+
+
+class ConnectContactFlowMediaStreams(DictWrapper):
+    @property
+    def customer(self) -> ConnectContactFlowMediaStreamCustomer:
+        return ConnectContactFlowMediaStreamCustomer(self["Customer"])
+
+
+class ConnectContactFlowData(DictWrapper):
+    @property
+    def attributes(self) -> Dict[str, str]:
+        """These are attributes that have been previously associated with a contact,
+        such as when using a Set contact attributes block in a contact flow.
+        This map may be empty if there aren't any saved attributes.
+        """
+        return self["Attributes"]
+
+    @property
+    def channel(self) -> ConnectContactFlowChannel:
+        """The method used to contact your contact center."""
+        return ConnectContactFlowChannel[self["Channel"]]
+
+    @property
+    def contact_id(self) -> str:
+        """The unique identifier of the contact."""
+        return self["ContactId"]
+
+    @property
+    def customer_endpoint(self) -> Optional[ConnectContactFlowEndpoint]:
+        """Contains the customer’s address (number) and type of address."""
+        if self["CustomerEndpoint"] is not None:
+            return ConnectContactFlowEndpoint(self["CustomerEndpoint"])
+        return None
+
+    @property
+    def initial_contact_id(self) -> str:
+        """The unique identifier for the contact associated with the first interaction between the customer and your
+        contact center. Use the initial contact ID to track contacts between contact flows.
+        """
+        return self["InitialContactId"]
+
+    @property
+    def initiation_method(self) -> ConnectContactFlowInitiationMethod:
+        """How the contact was initiated."""
+        return ConnectContactFlowInitiationMethod[self["InitiationMethod"]]
+
+    @property
+    def instance_arn(self) -> str:
+        """The ARN for your Amazon Connect instance."""
+        return self["InstanceARN"]
+
+    @property
+    def previous_contact_id(self) -> str:
+        """The unique identifier for the contact before it was transferred.
+        Use the previous contact ID to trace contacts between contact flows.
+        """
+        return self["PreviousContactId"]
+
+    @property
+    def queue(self) -> Optional[ConnectContactFlowQueue]:
+        """The current queue."""
+        if self["Queue"] is not None:
+            return ConnectContactFlowQueue(self["Queue"])
+        return None
+
+    @property
+    def system_endpoint(self) -> Optional[ConnectContactFlowEndpoint]:
+        """Contains the address (number) the customer dialed to call your contact center and type of address."""
+        if self["SystemEndpoint"] is not None:
+            return ConnectContactFlowEndpoint(self["SystemEndpoint"])
+        return None
+
+    @property
+    def media_streams(self) -> ConnectContactFlowMediaStreams:
+        return ConnectContactFlowMediaStreams(self["MediaStreams"])
+
+
+class ConnectContactFlowEvent(DictWrapper):
+    """Amazon Connect contact flow event
+
+    Documentation:
+    -------------
+    - https://docs.aws.amazon.com/connect/latest/adminguide/connect-lambda-functions.html
+    """
+
+    @property
+    def contact_data(self) -> ConnectContactFlowData:
+        """This is always passed by Amazon Connect for every contact. Some parameters are optional."""
+        return ConnectContactFlowData(self["Details"]["ContactData"])
+
+    @property
+    def parameters(self) -> Dict[str, str]:
+        """These are parameters specific to this call that were defined when you created the Lambda function."""
+        return self["Details"]["Parameters"]

tests/events/connectContactFlowEventAll.json

@@ -0,0 +1,41 @@
+{
+    "Name": "ContactFlowEvent",
+    "Details": {
+        "ContactData": {
+            "Attributes": {
+                "Language": "en-US"
+            },
+            "Channel": "VOICE",
+            "ContactId": "5ca32fbd-8f92-46af-92a5-6b0f970f0efe",
+            "CustomerEndpoint": {
+                "Address": "+11234567890",
+                "Type": "TELEPHONE_NUMBER"
+            },
+            "InitialContactId": "5ca32fbd-8f92-46af-92a5-6b0f970f0efe",
+            "InitiationMethod": "API",
+            "InstanceARN": "arn:aws:connect:eu-central-1:123456789012:instance/9308c2a1-9bc6-4cea-8290-6c0b4a6d38fa",
+            "MediaStreams": {
+                "Customer": {
+                    "Audio": {
+                        "StartFragmentNumber": "91343852333181432392682062622220590765191907586",
+                        "StartTimestamp": "1565781909613",
+                        "StreamARN": "arn:aws:kinesisvideo:eu-central-1:123456789012:stream/connect-contact-a3d73b84-ce0e-479a-a9dc-5637c9d30ac9/1565272947806"
+                    }
+                }
+            },
+            "PreviousContactId": "5ca32fbd-8f92-46af-92a5-6b0f970f0efe",
+            "Queue": {
+                "ARN": "arn:aws:connect:eu-central-1:123456789012:instance/9308c2a1-9bc6-4cea-8290-6c0b4a6d38fa/queue/5cba7cbf-1ecb-4b6d-b8bd-fe91079b3fc8",
+                "Name": "QueueOne"
+            },
+            "SystemEndpoint": {
+                "Address": "+11234567890",
+                "Type": "TELEPHONE_NUMBER"
+            }
+        },
+        "Parameters": {
+            "ParameterOne": "One",
+            "ParameterTwo": "Two"
+        }
+    }
+}

tests/events/connectContactFlowEventMin.json

@@ -0,0 +1,27 @@
+{
+    "Name": "ContactFlowEvent",
+    "Details": {
+        "ContactData": {
+            "Attributes": {},
+            "Channel": "VOICE",
+            "ContactId": "5ca32fbd-8f92-46af-92a5-6b0f970f0efe",
+            "CustomerEndpoint": null,
+            "InitialContactId": "5ca32fbd-8f92-46af-92a5-6b0f970f0efe",
+            "InitiationMethod": "API",
+            "InstanceARN": "arn:aws:connect:eu-central-1:123456789012:instance/9308c2a1-9bc6-4cea-8290-6c0b4a6d38fa",
+            "MediaStreams": {
+                "Customer": {
+                    "Audio": {
+                        "StartFragmentNumber": null,
+                        "StartTimestamp": null,
+                        "StreamARN": null
+                    }
+                }
+            },
+            "PreviousContactId": "5ca32fbd-8f92-46af-92a5-6b0f970f0efe",
+            "Queue": null,
+            "SystemEndpoint": null
+        },
+        "Parameters": {}
+    }
+}

tests/functional/test_lambda_trigger_events.py

@@ -29,6 +29,12 @@
     VerifyAuthChallengeResponseTriggerEvent,
 )
 from aws_lambda_powertools.utilities.data_classes.common import BaseProxyEvent, DictWrapper
+from aws_lambda_powertools.utilities.data_classes.connect_contact_flow_event import (
+    ConnectContactFlowChannel,
+    ConnectContactFlowEndpointType,
+    ConnectContactFlowEvent,
+    ConnectContactFlowInitiationMethod,
+)
 from aws_lambda_powertools.utilities.data_classes.dynamo_db_stream_event import (
     AttributeValue,
     DynamoDBRecordEventName,
@@ -318,6 +324,67 @@ def test_verify_auth_challenge_response_trigger_event():
     assert event.response.answer_correct is True
 
 
+def test_connect_contact_flow_event_min():
+    event = ConnectContactFlowEvent(load_event("connectContactFlowEventMin.json"))
+
+    assert event.contact_data.attributes == {}
+    assert event.contact_data.channel == ConnectContactFlowChannel.VOICE
+    assert event.contact_data.contact_id == "5ca32fbd-8f92-46af-92a5-6b0f970f0efe"
+    assert event.contact_data.customer_endpoint is None
+    assert event.contact_data.initial_contact_id == "5ca32fbd-8f92-46af-92a5-6b0f970f0efe"
+    assert event.contact_data.initiation_method == ConnectContactFlowInitiationMethod.API
+    assert (
+        event.contact_data.instance_arn
+        == "arn:aws:connect:eu-central-1:123456789012:instance/9308c2a1-9bc6-4cea-8290-6c0b4a6d38fa"
+    )
+    assert event.contact_data.media_streams.customer.audio.start_fragment_number is None
+    assert event.contact_data.media_streams.customer.audio.start_timestamp is None
+    assert event.contact_data.media_streams.customer.audio.stream_arn is None
+    assert event.contact_data.previous_contact_id == "5ca32fbd-8f92-46af-92a5-6b0f970f0efe"
+    assert event.contact_data.queue is None
+    assert event.contact_data.system_endpoint is None
+    assert event.parameters == {}
+
+
+def test_connect_contact_flow_event_all():
+    event = ConnectContactFlowEvent(load_event("connectContactFlowEventAll.json"))
+
+    assert event.contact_data.attributes == {"Language": "en-US"}
+    assert event.contact_data.channel == ConnectContactFlowChannel.VOICE
+    assert event.contact_data.contact_id == "5ca32fbd-8f92-46af-92a5-6b0f970f0efe"
+    assert event.contact_data.customer_endpoint is not None
+    assert event.contact_data.customer_endpoint.address == "+11234567890"
+    assert event.contact_data.customer_endpoint.endpoint_type == ConnectContactFlowEndpointType.TELEPHONE_NUMBER
+    assert event.contact_data.initial_contact_id == "5ca32fbd-8f92-46af-92a5-6b0f970f0efe"
+    assert event.contact_data.initiation_method == ConnectContactFlowInitiationMethod.API
+    assert (
+        event.contact_data.instance_arn
+        == "arn:aws:connect:eu-central-1:123456789012:instance/9308c2a1-9bc6-4cea-8290-6c0b4a6d38fa"
+    )
+    assert (
+        event.contact_data.media_streams.customer.audio.start_fragment_number
+        == "91343852333181432392682062622220590765191907586"
+    )
+    assert event.contact_data.media_streams.customer.audio.start_timestamp == "1565781909613"
+    assert (
+        event.contact_data.media_streams.customer.audio.stream_arn
+        == "arn:aws:kinesisvideo:eu-central-1:123456789012:stream/"
+        + "connect-contact-a3d73b84-ce0e-479a-a9dc-5637c9d30ac9/1565272947806"
+    )
+    assert event.contact_data.previous_contact_id == "5ca32fbd-8f92-46af-92a5-6b0f970f0efe"
+    assert event.contact_data.queue is not None
+    assert (
+        event.contact_data.queue.arn
+        == "arn:aws:connect:eu-central-1:123456789012:instance/9308c2a1-9bc6-4cea-8290-6c0b4a6d38fa/"
+        + "queue/5cba7cbf-1ecb-4b6d-b8bd-fe91079b3fc8"
+    )
+    assert event.contact_data.queue.name == "QueueOne"
+    assert event.contact_data.system_endpoint is not None
+    assert event.contact_data.system_endpoint.address == "+11234567890"
+    assert event.contact_data.system_endpoint.endpoint_type == ConnectContactFlowEndpointType.TELEPHONE_NUMBER
+    assert event.parameters == {"ParameterOne": "One", "ParameterTwo": "Two"}
+
+
 def test_dynamo_db_stream_trigger_event():
     event = DynamoDBStreamEvent(load_event("dynamoStreamEvent.json"))