feat(event_sources): add Secrets Manager secret rotation event (#3061)

Co-authored-by: Leandro Damascena <[email protected]>

Author: roger-zhangg

aws_lambda_powertools/utilities/data_classes/__init__.py

@@ -18,6 +18,7 @@
 from .kinesis_stream_event import KinesisStreamEvent
 from .lambda_function_url_event import LambdaFunctionUrlEvent
 from .s3_event import S3Event, S3EventBridgeNotificationEvent
+from .secrets_manager_event import SecretsManagerEvent
 from .ses_event import SESEvent
 from .sns_event import SNSEvent
 from .sqs_event import SQSEvent
@@ -26,6 +27,7 @@
 __all__ = [
     "APIGatewayProxyEvent",
     "APIGatewayProxyEventV2",
+    "SecretsManagerEvent",
     "AppSyncResolverEvent",
     "ALBEvent",
     "CloudWatchDashboardCustomWidgetEvent",

aws_lambda_powertools/utilities/data_classes/secrets_manager_event.py

@@ -0,0 +1,25 @@
+from typing_extensions import Literal
+
+from aws_lambda_powertools.utilities.data_classes.common import DictWrapper
+
+
+class SecretsManagerEvent(DictWrapper):
+    @property
+    def secret_id(self) -> str:
+        """SecretId: The secret ARN or identifier"""
+        return self["SecretId"]
+
+    @property
+    def client_request_token(self) -> str:
+        """ClientRequestToken: The ClientRequestToken associated with the secret version"""
+        return self["ClientRequestToken"]
+
+    @property
+    def version_id(self) -> str:
+        """Alias to ClientRequestToken to get token associated to version"""
+        return self["ClientRequestToken"]
+
+    @property
+    def step(self) -> Literal["createSecret", "setSecret", "testSecret", "finishSecret"]:
+        """Step: The rotation step (one of createSecret, setSecret, testSecret, or finishSecret)"""
+        return self["Step"]

docs/utilities/data_classes.md

@@ -1095,6 +1095,22 @@ This example is based on the AWS Blog post [Introducing Amazon S3 Object Lambda
         file_key = event.detail.object.key
     ```
 
+### Secrets Manager
+
+AWS Secrets Manager rotation uses an AWS Lambda function to update the secret. [Click here](https://docs.aws.amazon.com/secretsmanager/latest/userguide/rotating-secrets.html){target="_blank"} for more information about rotating AWS Secrets Manager secrets.
+
+=== "app.py"
+
+    ```python hl_lines="2 7 11"
+    --8<-- "examples/event_sources/src/secrets_manager.py"
+    ```
+
+=== "Secrets Manager Example Event"
+
+    ```json
+    --8<-- "tests/events/secretsManagerEvent.json"
+    ```
+
 ### SES
 
 === "app.py"

examples/event_sources/src/secrets_manager.py

@@ -0,0 +1,16 @@
+from aws_lambda_powertools.utilities import parameters
+from aws_lambda_powertools.utilities.data_classes import SecretsManagerEvent, event_source
+
+secrets_provider = parameters.SecretsProvider()
+
+
+@event_source(data_class=SecretsManagerEvent)
+def lambda_handler(event: SecretsManagerEvent, context):
+    # Getting secret value using Parameter utility
+    # See https://docs.powertools.aws.dev/lambda/python/latest/utilities/parameters/
+    secret = secrets_provider.get(event.secret_id, VersionId=event.version_id, VersionStage="AWSCURRENT")
+
+    # You need to work with secrets afterwards
+    # Check more examples: https://github.com/aws-samples/aws-secrets-manager-rotation-lambdas
+
+    return secret

examples/event_sources/src/secrets_manager_event.json

@@ -0,0 +1,5 @@
+{
+    "SecretId":"arn:aws:secretsmanager:us-west-2:123456789012:secret:MyTestDatabaseSecret-a1b2c3",
+    "ClientRequestToken":"550e8400-e29b-41d4-a716-446655440000",
+    "Step":"createSecret"
+}

tests/events/secretsManagerEvent.json

@@ -0,0 +1,5 @@
+{
+    "SecretId":"arn:aws:secretsmanager:us-west-2:123456789012:secret:MyTestDatabaseSecret-a1b2c3",
+    "ClientRequestToken":"550e8400-e29b-41d4-a716-446655440000",
+    "Step":"createSecret"
+}

tests/unit/data_classes/test_secrets_manager_event.py

@@ -0,0 +1,12 @@
+from aws_lambda_powertools.utilities.data_classes.secrets_manager_event import SecretsManagerEvent
+from tests.functional.utils import load_event
+
+
+def test_secrets_manager_event():
+    raw_event = load_event("secretsManagerEvent.json")
+    parsed_event = SecretsManagerEvent(raw_event)
+
+    assert parsed_event.secret_id == raw_event["SecretId"]
+    assert parsed_event.client_request_token == raw_event["ClientRequestToken"]
+    assert parsed_event.version_id == raw_event["ClientRequestToken"]
+    assert parsed_event.step == raw_event["Step"]