Merge branch 'develop' into refactoring-idempotent-ux

Author: Michael Brewer

aws_lambda_powertools/utilities/validation/jmespath_functions.py renamed to aws_lambda_powertools/shared/jmespath_functions.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/__init__.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"]

aws_lambda_powertools/utilities/data_classes/connect_contact_flow_event.py

@@ -1,8 +1,12 @@
+from typing import Dict
+
+
 class IdempotencyConfig:
     def __init__(
         self,
         event_key_jmespath: str = "",
         payload_validation_jmespath: str = "",
+        jmespath_options: Dict = None,
         raise_on_no_idempotency_key: bool = False,
         expires_after_seconds: int = 60 * 60,  # 1 hour default
         use_local_cache: bool = False,
@@ -31,6 +35,7 @@ def __init__(
         """
         self.event_key_jmespath = event_key_jmespath
         self.payload_validation_jmespath = payload_validation_jmespath
+        self.jmespath_options = jmespath_options
         self.raise_on_no_idempotency_key = raise_on_no_idempotency_key
         self.expires_after_seconds = expires_after_seconds
         self.use_local_cache = use_local_cache

aws_lambda_powertools/utilities/idempotency/config.py

@@ -14,6 +14,7 @@
 import jmespath
 
 from aws_lambda_powertools.shared.cache_dict import LRUDict
+from aws_lambda_powertools.shared.jmespath_functions import PowertoolsFunctions
 from aws_lambda_powertools.shared.json_encoder import Encoder
 from aws_lambda_powertools.utilities.idempotency.config import IdempotencyConfig
 from aws_lambda_powertools.utilities.idempotency.exceptions import (
@@ -112,6 +113,7 @@ def __init__(self):
 
         self.event_key_jmespath: Optional[str] = None
         self.event_key_compiled_jmespath = None
+        self.jmespath_options: Optional[dict] = None
         self.payload_validation_enabled = False
         self.validation_key_jmespath = None
         self.raise_on_no_idempotency_key = False
@@ -136,6 +138,9 @@ def configure(self, config: IdempotencyConfig,) -> None:
         self.event_key_jmespath = config.event_key_jmespath
         if config.event_key_jmespath:
             self.event_key_compiled_jmespath = jmespath.compile(config.event_key_jmespath)
+        self.jmespath_options = config.jmespath_options
+        if not self.jmespath_options:
+            self.jmespath_options = {"custom_functions": PowertoolsFunctions()}
         if config.payload_validation_jmespath:
             self.validation_key_jmespath = jmespath.compile(config.payload_validation_jmespath)
             self.payload_validation_enabled = True
@@ -162,8 +167,11 @@ def _get_hashed_idempotency_key(self, lambda_event: Dict[str, Any]) -> str:
 
         """
         data = lambda_event
+
         if self.event_key_jmespath:
-            data = self.event_key_compiled_jmespath.search(lambda_event)
+            data = self.event_key_compiled_jmespath.search(
+                lambda_event, options=jmespath.Options(**self.jmespath_options)
+            )
 
         if self.is_missing_idempotency_key(data):
             warnings.warn(f"No value found for idempotency_key. jmespath: {self.event_key_jmespath}")

aws_lambda_powertools/utilities/idempotency/persistence/base.py

@@ -5,8 +5,9 @@
 import jmespath
 from jmespath.exceptions import LexerError
 
+from aws_lambda_powertools.shared.jmespath_functions import PowertoolsFunctions
+
 from .exceptions import InvalidEnvelopeExpressionError, InvalidSchemaFormatError, SchemaValidationError
-from .jmespath_functions import PowertoolsFunctions
 
 logger = logging.getLogger(__name__)