feat(event_sources): add Secrets Manager secret rotation event

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_event import SecretManagerEvent
 from .ses_event import SESEvent
 from .sns_event import SNSEvent
 from .sqs_event import SQSEvent
@@ -26,6 +27,7 @@
 __all__ = [
     "APIGatewayProxyEvent",
     "APIGatewayProxyEventV2",
+    "SecretManagerEvent",
     "AppSyncResolverEvent",
     "ALBEvent",
     "CloudWatchDashboardCustomWidgetEvent",

aws_lambda_powertools/utilities/data_classes/secrets_event.py

@@ -0,0 +1,20 @@
+from typing_extensions import Literal
+
+from aws_lambda_powertools.utilities.data_classes.common import DictWrapper
+
+
+class SecretManagerEvent(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 of the secret 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

@@ -1161,6 +1161,26 @@ You can register your Lambda functions as targets within an Amazon VPC Lattice s
     --8<-- "examples/event_sources/src/vpc_lattice_payload.json"
     ```
 
+### Secrets Manager
+
+=== "app.py"
+
+    ```python
+    from aws_lambda_powertools.utilities.data_classes import event_source, SecretManagerEvent
+
+    @event_source(data_class=SecretManagerEvent)
+    def lambda_handler(event: SecretManagerEvent, context):
+        # Multiple records can be delivered in a single event
+        service_client = boto3.client('secretsmanager', endpoint_url=os.environ['SECRETS_MANAGER_ENDPOINT'])
+        create_secret(service_client, event.secret_id, event.client_request_token)
+        ...
+    ```
+=== "Secrets Manager Example Event"
+
+    ```json
+    --8<-- "tests/events/secretManagerEvent.json"
+    ```
+
 ## Advanced
 
 ### Debugging

tests/events/secretManagerEvent.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_secret_event.py

@@ -0,0 +1,11 @@
+from aws_lambda_powertools.utilities.data_classes.secrets_event import SecretManagerEvent
+from tests.functional.utils import load_event
+
+
+def test_vpc_lattice_event():
+    raw_event = load_event("secretManagerEvent.json")
+    parsed_event = SecretManagerEvent(raw_event)
+
+    assert parsed_event.secret_id == raw_event["SecretId"]
+    assert parsed_event.client_request_token == raw_event["ClientRequestToken"]
+    assert parsed_event.step == raw_event["Step"]