docs(data_classes): improve Event Source Data Classes documentation

examples/event_sources/events/active_mq_event_example.json

@@ -0,0 +1,27 @@
+{
+    "eventSource": "aws:mq",
+    "eventSourceArn": "arn:aws:mq:us-east-2:111122223333:broker:test:b-9bcfa592-423a-4942-879d-eb284b418fc8",
+    "messages": [
+        {
+            "messageID": "ID:b-9bcfa592-423a-4942-879d-eb284b418fc8-1.mq.us-east-2.amazonaws.com-37557-1234520418293-4:1:1:1:1",
+            "messageType": "jms/text-message",
+            "destination": {
+                "physicalName": "testQueue"
+            },
+            "data": "QUJDOkFBQUE=",
+            "timestamp": 1598827811958,
+            "properties": {
+                "index": "1"
+            }
+        },
+        {
+            "messageID": "ID:b-9bcfa592-423a-4942-879d-eb284b418fc8-1.mq.us-east-2.amazonaws.com-37557-1234520418293-4:1:1:1:2",
+            "messageType": "jms/bytes-message",
+            "destination": {
+                "physicalName": "testQueue2"
+            },
+            "data": "LQaGQ82S48k=",
+            "timestamp": 1598827811959
+        }
+    ]
+}

examples/event_sources/events/apigw_event.json

@@ -0,0 +1,20 @@
+{
+    "resource": "/helloworld",
+    "path": "/hello",
+    "httpMethod": "GET",
+    "headers": {
+        "Accept": "*/*",
+        "Host": "api.example.com"
+    },
+    "queryStringParameters": {
+        "name": "John"
+    },
+    "pathParameters": null,
+    "stageVariables": null,
+    "requestContext": {
+        "requestId": "c6af9ac6-7b61-11e6-9a41-93e8deadbeef",
+        "stage": "prod"
+    },
+    "body": null,
+    "isBase64Encoded": false
+}

examples/event_sources/events/s3ObjectEvent.json

@@ -0,0 +1,29 @@
+{
+    "xAmzRequestId": "1a5ed718-5f53-471d-b6fe-5cf62d88d02a",
+    "getObjectContext": {
+        "inputS3Url": "https://myap-123412341234.s3-accesspoint.us-east-1.amazonaws.com/s3.txt?X-Amz-Security-Token=...",
+        "outputRoute": "io-iad-cell001",
+        "outputToken": "..."
+    },
+    "configuration": {
+        "accessPointArn": "arn:aws:s3-object-lambda:us-east-1:123412341234:accesspoint/myolap",
+        "supportingAccessPointArn": "arn:aws:s3:us-east-1:123412341234:accesspoint/myap",
+        "payload": "test"
+    },
+    "userRequest": {
+        "url": "/s3.txt",
+        "headers": {
+            "Host": "myolap-123412341234.s3-object-lambda.us-east-1.amazonaws.com",
+            "Accept-Encoding": "identity",
+            "X-Amz-Content-SHA256": "e3b0c44297fc1c149afbf4c8995fb92427ae41e4649b934ca495991b7852b855"
+        }
+    },
+    "userIdentity": {
+        "type": "IAMUser",
+        "principalId": "...",
+        "arn": "arn:aws:iam::123412341234:user/myuser",
+        "accountId": "123412341234",
+        "accessKeyId": "..."
+    },
+    "protocolVersion": "1.00"
+}

examples/event_sources/src/active_mq_example.py

@@ -0,0 +1,18 @@
+import json
+
+from aws_lambda_powertools import Logger
+from aws_lambda_powertools.utilities.data_classes import event_source
+from aws_lambda_powertools.utilities.data_classes.active_mq_event import ActiveMQEvent
+
+logger = Logger()
+
+
+@event_source(data_class=ActiveMQEvent)
+def lambda_handler(event: ActiveMQEvent, context):
+    for message in event.messages:
+        msg = message.message_id
+        msg_pn = message.destination_physicalname
+
+        logger.info(f"Message ID: {msg} and physical name: {msg_pn}")
+
+    return {"statusCode": 200, "body": json.dumps("Processing complete")}

examples/event_sources/src/albEvent.py

@@ -0,0 +1,9 @@
+from aws_lambda_powertools.utilities.data_classes import ALBEvent, event_source
+
+
+@event_source(data_class=ALBEvent)
+def lambda_handler(event: ALBEvent, context):
+    if "lambda" in event.path and event.http_method == "GET":
+        return {"statusCode": 200, "body": f"Hello from path: {event.path}"}
+    else:
+        return {"statusCode": 400, "body": "No Hello from path"}

examples/event_sources/src/apigw_auth_v2.py

@@ -0,0 +1,30 @@
+from secrets import compare_digest
+
+from aws_lambda_powertools.utilities.data_classes import event_source
+from aws_lambda_powertools.utilities.data_classes.api_gateway_authorizer_event import (
+    APIGatewayAuthorizerEventV2,
+    APIGatewayAuthorizerResponseV2,
+)
+
+
+def get_user_by_token(token):
+    if compare_digest(token, "value"):
+        return {"name": "Foo"}
+    return None
+
+
+@event_source(data_class=APIGatewayAuthorizerEventV2)
+def lambda_handler(event: APIGatewayAuthorizerEventV2, context):
+    user = get_user_by_token(event.headers.get("Authorization"))
+
+    if user is None:
+        # No user was found, so we return not authorized
+        return APIGatewayAuthorizerResponseV2(authorize=False).asdict()
+
+    # Found the user and setting the details in the context
+    response = APIGatewayAuthorizerResponseV2(
+        authorize=True,
+        context=user,
+    )
+
+    return response.asdict()

examples/event_sources/src/apigw_authorizer_request.py

@@ -0,0 +1,29 @@
+from aws_lambda_powertools.utilities.data_classes import event_source
+from aws_lambda_powertools.utilities.data_classes.api_gateway_authorizer_event import (
+    APIGatewayAuthorizerRequestEvent,
+    APIGatewayAuthorizerResponse,
+)
+
+
+@event_source(data_class=APIGatewayAuthorizerRequestEvent)
+def lambda_handler(event: APIGatewayAuthorizerRequestEvent, context):
+    # Simple auth check (replace with your actual auth logic)
+    is_authorized = event.headers.get("HeaderAuth1") == "headerValue1"
+
+    if not is_authorized:
+        return {"principalId": "", "policyDocument": {"Version": "2012-10-17", "Statement": []}}
+
+    arn = event.parsed_arn
+
+    policy = APIGatewayAuthorizerResponse(
+        principal_id="user",
+        context={"user": "example"},
+        region=arn.region,
+        aws_account_id=arn.aws_account_id,
+        api_id=arn.api_id,
+        stage=arn.stage,
+    )
+
+    policy.allow_all_routes()
+
+    return policy.asdict()

examples/event_sources/src/apigw_authorizer_token.py

@@ -0,0 +1,29 @@
+from aws_lambda_powertools.utilities.data_classes import event_source
+from aws_lambda_powertools.utilities.data_classes.api_gateway_authorizer_event import (
+    APIGatewayAuthorizerResponse,
+    APIGatewayAuthorizerTokenEvent,
+)
+
+
+@event_source(data_class=APIGatewayAuthorizerTokenEvent)
+def lambda_handler(event: APIGatewayAuthorizerTokenEvent, context):
+    # Simple token check (replace with your actual token validation logic)
+    is_valid_token = event.authorization_token == "allow"
+
+    if not is_valid_token:
+        return {"principalId": "", "policyDocument": {"Version": "2012-10-17", "Statement": []}}
+
+    arn = event.parsed_arn
+
+    policy = APIGatewayAuthorizerResponse(
+        principal_id="user",
+        context={"user": "example"},
+        region=arn.region,
+        aws_account_id=arn.aws_account_id,
+        api_id=arn.api_id,
+        stage=arn.stage,
+    )
+
+    policy.allow_all_routes()
+
+    return policy.asdict()

examples/event_sources/src/apigw_proxy_decorator.py

@@ -0,0 +1,9 @@
+from aws_lambda_powertools.utilities.data_classes import APIGatewayProxyEvent, event_source
+
+
+@event_source(data_class=APIGatewayProxyEvent)
+def lambda_handler(event: APIGatewayProxyEvent, context):
+    if "hello" in event.path and event.http_method == "GET":
+        return {"statusCode": 200, "body": f"Hello from path: {event.path}"}
+    else:
+        return {"statusCode": 400, "body": "No Hello from path"}

examples/event_sources/src/apigw_proxy_v2.py

@@ -0,0 +1,9 @@
+from aws_lambda_powertools.utilities.data_classes import APIGatewayProxyEventV2, event_source
+
+
+@event_source(data_class=APIGatewayProxyEventV2)
+def lambda_handler(event: APIGatewayProxyEventV2, context):
+    if "hello" in event.path and event.http_method == "POST":
+        return {"statusCode": 200, "body": f"Hello from path: {event.path}"}
+    else:
+        return {"statusCode": 400, "body": "No Hello from path"}

examples/event_sources/src/appSyncAuthorizer.py

@@ -0,0 +1,33 @@
+from typing import Dict
+
+from aws_lambda_powertools.logging import correlation_paths
+from aws_lambda_powertools.logging.logger import Logger
+from aws_lambda_powertools.utilities.data_classes.appsync_authorizer_event import (
+    AppSyncAuthorizerEvent,
+    AppSyncAuthorizerResponse,
+)
+from aws_lambda_powertools.utilities.data_classes.event_source import event_source
+
+logger = Logger()
+
+
+def get_user_by_token(token: str):
+    """Look a user by token"""
+    ...
+
+
+@logger.inject_lambda_context(correlation_id_path=correlation_paths.APPSYNC_AUTHORIZER)
+@event_source(data_class=AppSyncAuthorizerEvent)
+def lambda_handler(event: AppSyncAuthorizerEvent, context) -> Dict:
+    user = get_user_by_token(event.authorization_token)
+
+    if not user:
+        # No user found, return not authorized
+        return AppSyncAuthorizerResponse().asdict()
+
+    return AppSyncAuthorizerResponse(
+        authorize=True,
+        resolver_context={"id": user.id},
+        # Only allow admins to delete events
+        deny_fields=None if user.is_admin else ["Mutation.deleteEvent"],
+    ).asdict()

examples/event_sources/src/appSyncResolver.py

@@ -0,0 +1,57 @@
+from aws_lambda_powertools.utilities.data_classes import event_source
+from aws_lambda_powertools.utilities.data_classes.appsync_resolver_event import (
+    AppSyncIdentityCognito,
+    AppSyncResolverEvent,
+)
+from aws_lambda_powertools.utilities.typing import LambdaContext
+
+
+@event_source(data_class=AppSyncResolverEvent)
+def lambda_handler(event: AppSyncResolverEvent, context: LambdaContext):
+    # Access the AppSync event details
+    type_name = event.type_name
+    field_name = event.field_name
+    arguments = event.arguments
+    source = event.source
+
+    print(f"Resolving field '{field_name}' for type '{type_name}'")
+    print(f"Arguments: {arguments}")
+    print(f"Source: {source}")
+
+    # Check if the identity is Cognito-based
+    if isinstance(event.identity, AppSyncIdentityCognito):
+        user_id = event.identity.sub
+        username = event.identity.username
+        print(f"Request from Cognito user: {username} (ID: {user_id})")
+    else:
+        print("Request is not from a Cognito-authenticated user")
+
+    if type_name == "Merchant" and field_name == "locations":
+        page = arguments.get("page", 1)
+        size = arguments.get("size", 10)
+        name_filter = arguments.get("name")
+
+        # Here you would typically fetch locations from a database
+        # This is a mock implementation
+        locations = [
+            {"id": "1", "name": "Location 1", "address": "123 Main St"},
+            {"id": "2", "name": "Location 2", "address": "456 Elm St"},
+            {"id": "3", "name": "Location 3", "address": "789 Oak St"},
+        ]
+
+        # Apply name filter if provided
+        if name_filter:
+            locations = [loc for loc in locations if name_filter.lower() in loc["name"].lower()]
+
+        # Apply pagination
+        start = (page - 1) * size
+        end = start + size
+        paginated_locations = locations[start:end]
+
+        return {
+            "items": paginated_locations,
+            "totalCount": len(locations),
+            "nextToken": str(page + 1) if end < len(locations) else None,
+        }
+    else:
+        raise Exception(f"Unhandled field: {field_name} for type: {type_name}")

examples/event_sources/src/aws_config_rule.py

@@ -3,13 +3,12 @@
     AWSConfigRuleEvent,
     event_source,
 )
-from aws_lambda_powertools.utilities.typing import LambdaContext
 
 logger = Logger()
 
 
 @event_source(data_class=AWSConfigRuleEvent)
-def lambda_handler(event: AWSConfigRuleEvent, context: LambdaContext):
+def lambda_handler(event: AWSConfigRuleEvent, context):
     message_type = event.invoking_event.message_type
 
     logger.info(f"Logging {message_type} event rule", invoke_event=event.raw_invoking_event)