feat(event_sources): add support for Amazon Connect Outbound Campaigns

aws_lambda_powertools/utilities/data_classes/__init__.py

@@ -25,6 +25,13 @@
     CodeDeployLifecycleHookEvent,
 )
 from .code_pipeline_job_event import CodePipelineJobEvent
+from .connect_campaign_event import (
+    ConnectOutboundCampaignContext,
+    ConnectOutboundCampaignCustomerProfile,
+    ConnectOutboundCampaignInvocationMetadata,
+    ConnectOutboundCampaignItems,
+    ConnectOutboundCampaignsEvent,
+)
 from .connect_contact_flow_event import ConnectContactFlowEvent
 from .dynamo_db_stream_event import DynamoDBStreamEvent
 from .event_bridge_event import EventBridgeEvent
@@ -73,6 +80,11 @@
     "CodeDeployLifecycleHookEvent",
     "CodePipelineJobEvent",
     "ConnectContactFlowEvent",
+    "ConnectOutboundCampaignsEvent",
+    "ConnectOutboundCampaignInvocationMetadata",
+    "ConnectOutboundCampaignContext",
+    "ConnectOutboundCampaignItems",
+    "ConnectOutboundCampaignCustomerProfile",
     "DynamoDBStreamEvent",
     "EventBridgeEvent",
     "EventBridgeV2Event",

aws_lambda_powertools/utilities/data_classes/connect_campaign_event.py

@@ -0,0 +1,100 @@
+from __future__ import annotations
+
+from functools import cached_property
+from typing import TYPE_CHECKING, Any
+
+from aws_lambda_powertools.utilities.data_classes.common import DictWrapper
+
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+
+
+class ConnectOutboundCampaignContext(DictWrapper):
+    """Campaign and execution context details."""
+
+    @property
+    def campaign_id(self) -> str:
+        """The identifier of the campaign."""
+        return self["CampaignId"]
+
+    @property
+    def run_id(self) -> str:
+        """The identifier of the campaign run."""
+        return self["RunId"]
+
+    @property
+    def action_id(self) -> str:
+        """The identifier of the activity or action block."""
+        return self["ActionId"]
+
+    @property
+    def campaign_name(self) -> str:
+        """The name of the campaign."""
+        return self["CampaignName"]
+
+
+class ConnectOutboundCampaignInvocationMetadata(DictWrapper):
+    """Invocation metadata containing the campaign context."""
+
+    @property
+    def campaign_context(self) -> ConnectOutboundCampaignContext:
+        """Metadata for the campaign and execution context."""
+        return ConnectOutboundCampaignContext(self["CampaignContext"])
+
+
+class ConnectOutboundCampaignCustomerProfile(DictWrapper):
+    """Customer profile record within an outbound campaign batch."""
+
+    @property
+    def profile_id(self) -> str:
+        """The identifier of the customer profile."""
+        return self["ProfileId"]
+
+    @property
+    def customer_data(self) -> str:
+        """The JSON-encoded customer data string configured in the campaign."""
+        return self["CustomerData"]
+
+    @cached_property
+    def json_customer_data(self) -> Any:
+        """Parsed JSON object of customer data."""
+        return self._json_deserializer(self["CustomerData"])
+
+    @property
+    def idempotency_token(self) -> str:
+        """The idempotency token for deduplication."""
+        return self["IdempotencyToken"]
+
+
+class ConnectOutboundCampaignItems(DictWrapper):
+    """Container for customer profile items."""
+
+    @property
+    def customer_profiles(self) -> Iterator[ConnectOutboundCampaignCustomerProfile]:
+        """Iterate over customer profiles."""
+        for item in self["CustomerProfiles"]:
+            yield ConnectOutboundCampaignCustomerProfile(item)
+
+
+class ConnectOutboundCampaignsEvent(DictWrapper):
+    """Amazon Connect Outbound Campaigns custom action event.
+
+    Documentation:
+    --------------
+    - https://docs.aws.amazon.com/connect/latest/adminguide/lambda-invoke-functions.html
+    """
+
+    @property
+    def invocation_metadata(self) -> ConnectOutboundCampaignInvocationMetadata:
+        """Invocation metadata including campaign context."""
+        return ConnectOutboundCampaignInvocationMetadata(self["InvocationMetadata"])
+
+    @property
+    def items_payload(self) -> ConnectOutboundCampaignItems:
+        """Items payload containing customer profiles."""
+        return ConnectOutboundCampaignItems(self["Items"])
+
+    @property
+    def customer_profiles(self) -> Iterator[ConnectOutboundCampaignCustomerProfile]:
+        """Convenience accessor to iterate directly over customer profiles."""
+        return self.items_payload.customer_profiles

aws_lambda_powertools/utilities/parser/models/__init__.py

@@ -70,6 +70,13 @@
         CognitoPreTokenGenerationTriggerModelV2AndV3,
         CognitoVerifyAuthChallengeTriggerModel,
     )
+    from .connect_campaign import (
+        ConnectCampaignContextModel,
+        ConnectCustomerProfileModel,
+        ConnectInvocationMetadataModel,
+        ConnectItemsModel,
+        ConnectOutboundCampaignsEventModel,
+    )
     from .dynamodb import (
         DynamoDBStreamChangedRecordModel,
         DynamoDBStreamLambdaOnFailureDestinationModel,
@@ -182,6 +189,11 @@
     "CloudWatchLogsDecode": "cloudwatch",
     "CloudWatchLogsLogEvent": "cloudwatch",
     "CloudWatchLogsModel": "cloudwatch",
+    "ConnectCampaignContextModel": "connect_campaign",
+    "ConnectCustomerProfileModel": "connect_campaign",
+    "ConnectInvocationMetadataModel": "connect_campaign",
+    "ConnectItemsModel": "connect_campaign",
+    "ConnectOutboundCampaignsEventModel": "connect_campaign",
     "CognitoCreateAuthChallengeTriggerModel": "cognito",
     "CognitoCustomEmailSenderTriggerModel": "cognito",
     "CognitoCustomMessageTriggerModel": "cognito",
@@ -295,6 +307,11 @@ def __dir__() -> list[str]:
     "CloudWatchLogsDecode",
     "CloudWatchLogsLogEvent",
     "CloudWatchLogsModel",
+    "ConnectCampaignContextModel",
+    "ConnectCustomerProfileModel",
+    "ConnectInvocationMetadataModel",
+    "ConnectItemsModel",
+    "ConnectOutboundCampaignsEventModel",
     "CognitoPreSignupTriggerModel",
     "CognitoPostConfirmationTriggerModel",
     "CognitoPreAuthenticationTriggerModel",

aws_lambda_powertools/utilities/parser/models/connect_campaign.py

@@ -0,0 +1,50 @@
+from __future__ import annotations
+
+from typing import Any
+
+from pydantic import BaseModel, Field
+
+
+class ConnectCampaignContextModel(BaseModel):
+    """Campaign context details model."""
+
+    campaign_id: str = Field(..., alias="CampaignId")
+    run_id: str = Field(..., alias="RunId")
+    action_id: str = Field(..., alias="ActionId")
+    campaign_name: str = Field(..., alias="CampaignName")
+
+
+class ConnectInvocationMetadataModel(BaseModel):
+    """Invocation metadata container model."""
+
+    campaign_context: ConnectCampaignContextModel = Field(..., alias="CampaignContext")
+
+
+class ConnectCustomerProfileModel(BaseModel):
+    """Customer profile record model.
+
+    CustomerData is campaign-defined and can be a raw JSON string or structured object.
+    Users can inherit from this model to provide a typed schema for CustomerData.
+    """
+
+    profile_id: str = Field(..., alias="ProfileId")
+    customer_data: Any = Field(..., alias="CustomerData")
+    idempotency_token: str = Field(..., alias="IdempotencyToken")
+
+
+class ConnectItemsModel(BaseModel):
+    """Container for customer profiles."""
+
+    customer_profiles: list[ConnectCustomerProfileModel] = Field(..., alias="CustomerProfiles")
+
+
+class ConnectOutboundCampaignsEventModel(BaseModel):
+    """Amazon Connect Outbound Campaigns custom action event model.
+
+    Documentation:
+    --------------
+    - https://docs.aws.amazon.com/connect/latest/adminguide/lambda-invoke-functions.html
+    """
+
+    invocation_metadata: ConnectInvocationMetadataModel = Field(..., alias="InvocationMetadata")
+    items: ConnectItemsModel = Field(..., alias="Items")

docs/utilities/data_classes.md

@@ -85,6 +85,7 @@ Each event source is linked to its corresponding GitHub file with the full set o
 | [CodePipeline Job Event](#codepipeline-job) | `CodePipelineJobEvent` | [Github](https://github.com/aws-powertools/powertools-lambda-python/blob/develop/aws_lambda_powertools/utilities/data_classes/code_pipeline_job_event.py) |
 | [Cognito User Pool](#cognito-user-pool) | Multiple available under `cognito_user_pool_event` | [Github](https://github.com/aws-powertools/powertools-lambda-python/blob/develop/aws_lambda_powertools/utilities/data_classes/cognito_user_pool_event.py) |
 | [Connect Contact Flow](#connect-contact-flow) | `ConnectContactFlowEvent` | [Github](https://github.com/aws-powertools/powertools-lambda-python/blob/develop/aws_lambda_powertools/utilities/data_classes/connect_contact_flow_event.py) |
+| [Connect Outbound Campaigns](#connect-outbound-campaigns) | `ConnectOutboundCampaignsEvent` | [Github](https://github.com/aws-powertools/powertools-lambda-python/blob/develop/aws_lambda_powertools/utilities/data_classes/connect_campaign_event.py) |
 | [DynamoDB streams](#dynamodb-streams) | `DynamoDBStreamEvent`, `DynamoDBRecordEventName` | [Github](https://github.com/aws-powertools/powertools-lambda-python/blob/develop/aws_lambda_powertools/utilities/data_classes/dynamo_db_stream_event.py) |
 | [EventBridge](#eventbridge) | `EventBridgeEvent` | [Github](https://github.com/aws-powertools/powertools-lambda-python/blob/develop/aws_lambda_powertools/utilities/data_classes/event_bridge_event.py) |
 | [EventBridge Custom Event Bus](#eventbridge-custom-event-bus) | `EventBridgeV2Event` | [Github](https://github.com/aws-powertools/powertools-lambda-python/blob/develop/aws_lambda_powertools/utilities/data_classes/event_bridge_v2_event.py) |
@@ -519,6 +520,21 @@ The example integrates with [Amazon Connect](https://docs.aws.amazon.com/connect
     --8<-- "tests/events/connectContactFlowEventAll.json"
     ```
 
+### Connect Outbound Campaigns
+
+The example integrates with [Amazon Connect Outbound Campaigns](https://docs.aws.amazon.com/connect/latest/adminguide/lambda-invoke-functions.html) by handling custom action block events in journey flows. The function converts the event into a `ConnectOutboundCampaignsEvent` object.
+
+=== "app.py"
+
+    ```python hl_lines="1-5 9"
+    --8<-- "examples/event_sources/src/connect_outbound_campaigns.py"
+    ```
+
+=== "Connect Outbound Campaigns Example Event"
+    ```json
+    --8<-- "tests/events/connectOutboundCampaignsEvent.json"
+    ```
+
 ### DynamoDB Streams
 
 The DynamoDB data class utility provides the base class for `DynamoDBStreamEvent`, as well as enums for stream view type (`StreamViewType`) and event type.

docs/utilities/parser.md

@@ -119,6 +119,7 @@ The example above uses `SqsModel`. Other built-in models can be found below.
 | **CloudFormationCustomResourceUpdateModel**       | Lambda Event Source payload for AWS CloudFormation `UPDATE` operation                         |
 | **CloudFormationCustomResourceDeleteModel**       | Lambda Event Source payload for AWS CloudFormation `DELETE` operation                         |
 | **CloudwatchLogsModel**                           | Lambda Event Source payload for Amazon CloudWatch Logs                                        |
+| **ConnectOutboundCampaignsEventModel**            | Lambda Event Source payload for Amazon Connect Outbound Campaigns custom actions              |
 | **CognitoPreSignupTriggerModel**                  | Lambda User Pool Pre-Sign-Up trigger event                                                    |
 | **CognitoPostConfirmationTriggerModel**           | Lambda User Pool Post Confirmation trigger event                                              |
 | **CognitoPreAuthenticationTriggerModel**          | Lambda User Pool Pre Authentication trigger event                                             |

examples/event_sources/src/connect_outbound_campaigns.py

@@ -0,0 +1,17 @@
+from aws_lambda_powertools.utilities.data_classes import (
+    ConnectOutboundCampaignsEvent,
+    event_source,
+)
+from aws_lambda_powertools.utilities.typing import LambdaContext
+
+
+@event_source(data_class=ConnectOutboundCampaignsEvent)
+def lambda_handler(event: ConnectOutboundCampaignsEvent, context: LambdaContext):
+    campaign = event.invocation_metadata.campaign_context
+    print(f"Processing campaign {campaign.campaign_name} (ID: {campaign.campaign_id})")
+
+    for profile in event.customer_profiles:
+        profile_id = profile.profile_id
+        # json_customer_data deserializes the embedded JSON string
+        customer_info = profile.json_customer_data
+        print(f"Customer {profile_id}: {customer_info}")

tests/events/connectOutboundCampaignsEvent.json

@@ -0,0 +1,24 @@
+{
+  "InvocationMetadata": {
+    "CampaignContext": {
+      "CampaignId": "campaign-12345",
+      "RunId": "run-67890",
+      "ActionId": "activity-abc123",
+      "CampaignName": "Welcome Campaign"
+    }
+  },
+  "Items": {
+    "CustomerProfiles": [
+      {
+        "ProfileId": "customer-001",
+        "CustomerData": "{\"firstName\":\"John\",\"lastName\":\"Doe\",\"email\":\"john.doe@example.com\"}",
+        "IdempotencyToken": "token-xyz789"
+      },
+      {
+        "ProfileId": "customer-002",
+        "CustomerData": "{\"firstName\":\"Jane\",\"lastName\":\"Smith\",\"email\":\"jane.smith@example.com\"}",
+        "IdempotencyToken": "token-abc456"
+      }
+    ]
+  }
+}

tests/unit/data_classes/required_dependencies/test_connect_campaign_event.py

@@ -0,0 +1,45 @@
+from aws_lambda_powertools.utilities.data_classes import ConnectOutboundCampaignsEvent
+from tests.functional.utils import load_event
+
+
+def test_connect_outbound_campaigns_event():
+    raw_event = load_event("connectOutboundCampaignsEvent.json")
+    event = ConnectOutboundCampaignsEvent(raw_event)
+
+    # Invocation metadata
+    invoc_meta = event.invocation_metadata
+    campaign_context = invoc_meta.campaign_context
+
+    assert campaign_context.campaign_id == "campaign-12345"
+    assert campaign_context.run_id == "run-67890"
+    assert campaign_context.action_id == "activity-abc123"
+    assert campaign_context.campaign_name == "Welcome Campaign"
+
+    # Items and customer profiles via items_payload
+    profiles = list(event.items_payload.customer_profiles)
+    assert len(profiles) == 2
+
+    # Direct convenience accessor
+    assert len(list(event.customer_profiles)) == 2
+
+    # First profile
+    p1 = profiles[0]
+    assert p1.profile_id == "customer-001"
+    assert p1.customer_data == '{"firstName":"John","lastName":"Doe","email":"john.doe@example.com"}'
+    assert p1.json_customer_data == {
+        "firstName": "John",
+        "lastName": "Doe",
+        "email": "john.doe@example.com",
+    }
+    assert p1.idempotency_token == "token-xyz789"
+
+    # Second profile
+    p2 = profiles[1]
+    assert p2.profile_id == "customer-002"
+    assert p2.customer_data == '{"firstName":"Jane","lastName":"Smith","email":"jane.smith@example.com"}'
+    assert p2.json_customer_data == {
+        "firstName": "Jane",
+        "lastName": "Smith",
+        "email": "jane.smith@example.com",
+    }
+    assert p2.idempotency_token == "token-abc456"