Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat(event_source): add CodeDeploy Lifecycle Hook event #5219

4 changes: 4 additions & 0 deletions aws_lambda_powertools/utilities/data_classes/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,9 @@
from .cloud_watch_custom_widget_event import CloudWatchDashboardCustomWidgetEvent
from .cloud_watch_logs_event import CloudWatchLogsEvent
from .cloudformation_custom_resource_event import CloudFormationCustomResourceEvent
from .code_deploy_lifecycle_hook_event import (
CodeDeployLifecycleHookEvent,
)
from .code_pipeline_job_event import CodePipelineJobEvent
from .connect_contact_flow_event import ConnectContactFlowEvent
from .dynamo_db_stream_event import DynamoDBStreamEvent
Expand Down Expand Up @@ -59,6 +62,7 @@
"CloudWatchAlarmMetricStat",
"CloudWatchDashboardCustomWidgetEvent",
"CloudWatchLogsEvent",
"CodeDeployLifecycleHookEvent",
"CodePipelineJobEvent",
"ConnectContactFlowEvent",
"DynamoDBStreamEvent",
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,13 @@
from aws_lambda_powertools.utilities.data_classes.common import DictWrapper


class CodeDeployLifecycleHookEvent(DictWrapper):
@property
def deployment_id(self) -> str:
"""The unique ID of the calling CodeDeploy Deployment."""
return self["DeploymentId"]

@property
def lifecycle_event_hook_execution_id(self) -> str:
"""The unique ID of a deployments lifecycle hook."""
return self["LifecycleEventHookExecutionId"]
25 changes: 25 additions & 0 deletions docs/utilities/data_classes.md
Original file line number Diff line number Diff line change
Expand Up @@ -90,6 +90,7 @@ Log Data Event for Troubleshooting
| [CloudWatch Alarm State Change Action](#cloudwatch-alarm-state-change-action) | `CloudWatchAlarmEvent` |
| [CloudWatch Dashboard Custom Widget](#cloudwatch-dashboard-custom-widget) | `CloudWatchDashboardCustomWidgetEvent` |
| [CloudWatch Logs](#cloudwatch-logs) | `CloudWatchLogsEvent` |
| [CodeDeploy Lifecycle Hook](#codedeploy-lifecycle-hook) | `CodeDeployLifecycleHookEvent` |
| [CodePipeline Job Event](#codepipeline-job) | `CodePipelineJobEvent` |
| [Cognito User Pool](#cognito-user-pool) | Multiple available under `cognito_user_pool_event` |
| [Connect Contact Flow](#connect-contact-flow) | `ConnectContactFlowEvent` |
Expand Down Expand Up @@ -615,6 +616,30 @@ Alternatively, you can use `extract_cloudwatch_logs_from_record` to seamless int
return processor.response()
```

### CodeDeploy LifeCycle Hook

CodeDeploy triggers Lambdas with this event when defined in
[AppSpec definitions](https://docs.aws.amazon.com/codedeploy/latest/userguide/reference-appspec-file-structure-hooks.html)
to test applications at different stages of deployment.


=== "app.py"
```python
from aws_lambda_powertools import Logger
from aws_lambda_powertools.utilities.data_classes import (
event_source,
CodeDeployLifecycleHookEvent,
)

logger = Logger()

def lambda_handler(
event: CodeDeployLifecycleHookEvent, context: LambdaContext
) -> None:
deployment_id = event.deployment_id
lifecycle_event_hook_execution_id = event.lifecycle_event_hook_execution_id
```

### CodePipeline Job

Data classes and utility functions to help create continuous delivery pipelines tasks with AWS Lambda
Expand Down
4 changes: 4 additions & 0 deletions tests/events/codeDeployLifecycleHookEvent.json
Original file line number Diff line number Diff line change
@@ -0,0 +1,4 @@
{
"DeploymentId": "d-ABCDEF",
"LifecycleEventHookExecutionId": "xxxxxxxxxxxxxxxxxxxxxxxx"
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,20 @@
import pytest

from aws_lambda_powertools.utilities.data_classes import (
CodeDeployLifecycleHookEvent,
)
from tests.functional.utils import load_event


@pytest.mark.parametrize(
"event_file",
[
"codeDeployLifecycleHookEvent.json",
],
)
def test_code_deploy_lifecycle_hook_event(event_file):
raw_event = load_event(event_file)
parsed_event = CodeDeployLifecycleHookEvent(raw_event)

assert parsed_event.deployment_id == raw_event["DeploymentId"]
assert parsed_event.lifecycle_event_hook_execution_id == raw_event["LifecycleEventHookExecutionId"]