aws-durable-execution-sdk-python/packages/aws-durable-execution-sdk-python-examples/scripts/generate_sam_template.py at 9589584a4c547542790db9181ce4474384e767cd · aws/aws-durable-execution-sdk-python · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
#!/usr/bin/env python3

import json
from pathlib import Path


def load_catalog():
    """Load examples catalog."""
    catalog_path = Path(__file__).parent.parent / "examples-catalog.json"
    with open(catalog_path) as f:
        return json.load(f)


def generate_sam_template():
    """Generate SAM template for all examples."""
    catalog = load_catalog()

    template = {
        "AWSTemplateFormatVersion": "2010-09-09",
        "Transform": "AWS::Serverless-2016-10-31",
        "Globals": {
            "Function": {
                "Runtime": "python3.13",
                "Timeout": 60,
                "MemorySize": 128,
                "Environment": {
                    "Variables": {"AWS_ENDPOINT_URL_LAMBDA": {"Ref": "LambdaEndpoint"}}
                },
            }
        },
        "Parameters": {
            "LambdaEndpoint": {
                "Type": "String",
                "Default": "https://lambda.us-west-2.amazonaws.com",
            }
        },
        "Resources": {
            "DurableFunctionRole": {
                "Type": "AWS::IAM::Role",
                "Properties": {
                    "AssumeRolePolicyDocument": {
                        "Version": "2012-10-17",
                        "Statement": [
                            {
                                "Effect": "Allow",
                                "Principal": {"Service": "lambda.amazonaws.com"},
                                "Action": "sts:AssumeRole",
                            }
                        ],
                    },
                    "ManagedPolicyArns": [
                        "arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole"
                    ],
                    "Policies": [
                        {
                            "PolicyName": "DurableExecutionPolicy",
                            "PolicyDocument": {
                                "Version": "2012-10-17",
                                "Statement": [
                                    {
                                        "Effect": "Allow",
                                        "Action": [
                                            "lambda:CheckpointDurableExecution",
                                            "lambda:GetDurableExecutionState",
                                        ],
                                        "Resource": "*",
                                    }
                                ],
                            },
                        }
                    ],
                },
            }
        },
    }

    for example in catalog["examples"]:
        # Convert handler name to PascalCase (e.g., hello_world -> HelloWorld)
        handler_base = example["handler"].replace(".handler", "")
        function_name = "".join(word.capitalize() for word in handler_base.split("_"))
        template["Resources"][function_name] = {
            "Type": "AWS::Serverless::Function",
            "Properties": {
                "CodeUri": "build/",
                "Handler": example["handler"],
                "Description": example["description"],
                "Role": {"Fn::GetAtt": ["DurableFunctionRole", "Arn"]},
            },
        }

        if "durableConfig" in example:
            template["Resources"][function_name]["Properties"]["DurableConfig"] = (
                example["durableConfig"]
            )

        if "layers" in example:
            template["Resources"][function_name]["Properties"]["Layers"] = example[
                "layers"
            ]

        if "tracing" in example:
            template["Resources"][function_name]["Properties"]["Tracing"] = example[
                "tracing"
            ]

        if "environment" in example:
            template["Resources"][function_name]["Properties"]["Environment"] = {
                "Variables": example["environment"]
            }

    template_path = Path(__file__).parent.parent / "template.yaml"
    with open(template_path, "w") as f:
        json.dump(template, f, sort_keys=False, indent=2)

    print(f"Generated SAM template at {template_path}")


if __name__ == "__main__":
    generate_sam_template()