Discord Powertools for AWS Lambda - Parsing EventBridge ECS Task State Change event

app.py

from __future__ import annotations

from typing import List, Optional

from aws_lambda_powertools.utilities.parser.models import EventBridgeModel
from aws_lambda_powertools.utilities.parser.envelopes import (
    EventBridgeEnvelope,  # only if you want to discard EventBridge metadata
)
from aws_lambda_powertools.utilities.parser.parser import event_parser
from pydantic import BaseModel


class Detail(BaseModel):
    name: str
    value: str


class Attachment(BaseModel):
    id: str
    type: str
    status: str
    details: List[Detail]


class NetworkInterface(BaseModel):
    attachmentId: str
    privateIpv4Address: str


class Container(BaseModel):
    containerArn: str
    lastStatus: str
    name: str
    image: str
    imageDigest: Optional[str]
    runtimeId: str
    taskArn: str
    networkInterfaces: List[NetworkInterface]
    cpu: str


class ContainerOverride(BaseModel):
    name: str


class Overrides(BaseModel):
    containerOverrides: List[ContainerOverride]


class ECSTaskStateChangeEvent(BaseModel):
    attachments: List[Attachment]
    availabilityZone: str
    clusterArn: str
    containers: List[Container]
    createdAt: str
    launchType: str
    cpu: str
    memory: str
    desiredStatus: str
    group: str
    lastStatus: str
    overrides: Overrides
    connectivity: str
    connectivityAt: str
    pullStartedAt: Optional[str]
    startedAt: str
    pullStoppedAt: str
    updatedAt: str
    taskArn: str
    taskDefinitionArn: str
    version: int
    platformVersion: str


class ECSTaskStateChangeEventBridge(EventBridgeModel):
    detail: ECSTaskStateChangeEvent


# @event_parser uses type annotation to infer which model it should try to parse and validate
@event_parser
def lambda_handler(event: ECSTaskStateChangeEventBridge, _):
    print(event.detail.clusterArn)


# `envelope` feature discards the EventBridge wrapping, leaving only your actual ECS Task event
@event_parser(model=ECSTaskStateChangeEvent, envelope=EventBridgeEnvelope)
def lambda_handler_without_eventbridge_metadata(event: ECSTaskStateChangeEvent, _):
    print(event.clusterArn)


sample_event = {
    "version": "0",
    "id": "3317b2af-7005-947d-b652-f55e762e571a",
    "detail-type": "ECS Task State Change",
    "source": "aws.ecs",
    "account": "111122223333",
    "time": "2020-01-23T17:57:58Z",
    "region": "us-west-2",
    "resources": ["arn:aws:ecs:us-west-2:111122223333:task/FargateCluster/c13b4cb40f1f4fe4a2971f76ae5a47ad"],
    "detail": {
        "attachments": [
            {
                "id": "1789bcae-ddfb-4d10-8ebe-8ac87ddba5b8",
                "type": "eni",
                "status": "ATTACHED",
                "details": [
                    {"name": "subnetId", "value": "subnet-abcd1234"},
                    {"name": "networkInterfaceId", "value": "eni-abcd1234"},
                    {"name": "macAddress", "value": "0a:98:eb:a7:29:ba"},
                    {"name": "privateIPv4Address", "value": "10.0.0.139"},
                ],
            }
        ],
        "availabilityZone": "us-west-2c",
        "clusterArn": "arn:aws:ecs:us-west-2:111122223333:cluster/FargateCluster",
        "containers": [
            {
                "containerArn": "arn:aws:ecs:us-west-2:111122223333:container/cf159fd6-3e3f-4a9e-84f9-66cbe726af01",
                "lastStatus": "RUNNING",
                "name": "FargateApp",
                "image": "111122223333.dkr.ecr.us-west-2.amazonaws.com/hello-repository:latest",
                "imageDigest": "sha256:74b2c688c700ec95a93e478cdb959737c148df3fbf5ea706abe0318726e885e6",
                "runtimeId": "ad64cbc71c7fb31c55507ec24c9f77947132b03d48d9961115cf24f3b7307e1e",
                "taskArn": "arn:aws:ecs:us-west-2:111122223333:task/FargateCluster/c13b4cb40f1f4fe4a2971f76ae5a47ad",
                "networkInterfaces": [
                    {"attachmentId": "1789bcae-ddfb-4d10-8ebe-8ac87ddba5b8", "privateIpv4Address": "10.0.0.139"}
                ],
                "cpu": "0",
            }
        ],
        "createdAt": "2020-01-23T17:57:34.402Z",
        "launchType": "FARGATE",
        "cpu": "256",
        "memory": "512",
        "desiredStatus": "RUNNING",
        "group": "family:sample-fargate",
        "lastStatus": "RUNNING",
        "overrides": {"containerOverrides": [{"name": "FargateApp"}]},
        "connectivity": "CONNECTED",
        "connectivityAt": "2020-01-23T17:57:38.453Z",
        "pullStartedAt": "2020-01-23T17:57:52.103Z",
        "startedAt": "2020-01-23T17:57:58.103Z",
        "pullStoppedAt": "2020-01-23T17:57:55.103Z",
        "updatedAt": "2020-01-23T17:57:58.103Z",
        "taskArn": "arn:aws:ecs:us-west-2:111122223333:task/FargateCluster/c13b4cb40f1f4fe4a2971f76ae5a47ad",
        "taskDefinitionArn": "arn:aws:ecs:us-west-2:111122223333:task-definition/sample-fargate:1",
        "version": 4,
        "platformVersion": "1.3.0",
    },
}

# what most customers prefer since you can use EventBridge metadata info for observability
lambda_handler(sample_event, {})

lambda_handler_without_eventbridge_metadata(sample_event, {})