Coverage for scripts/generate_sam_template.py: 99%
116 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-07-07 16:54 +0000
1#!/usr/bin/env python3
3import argparse
4import ast
5import json
6import sys
7from pathlib import Path
8from typing import Any
10if not __package__:
11 sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
13from scripts.function_naming import to_function_name_suffix, to_logical_id
14from scripts.test_handlers import load_test_handlers
16PACKAGE_NAME = "DurableExecutionsPythonExamples-1.0"
17PACKAGE_PREFIX = "async_durable_execution_examples"
18DEFAULT_AWS_REGION = "eu-south-1"
19DEFAULT_LAMBDA_ENDPOINT = f"https://lambda.{DEFAULT_AWS_REGION}.amazonaws.com"
20DEFAULT_RUNTIME = "python3.13"
21SDK_LAYER_LOGICAL_ID = "AsyncDurableExecutionSdkLayer"
22SDK_LAYER_CONTENT_URI = "../dist/async-durable-execution-layer.zip"
23DEFAULT_DURABLE_CONFIG = {
24 "RetentionPeriodInDays": 7,
25 "ExecutionTimeout": 300,
26}
27SPECIAL_LOGGING_CONFIG = {
28 "callback/callback_concurrency.py": {
29 "ApplicationLogLevel": "DEBUG",
30 "LogFormat": "JSON",
31 },
32 "logger_example/logger_example.py": {
33 "ApplicationLogLevel": "INFO",
34 "LogFormat": "JSON",
35 },
36}
39def build_examples_catalog() -> dict[str, Any]:
40 """Build the examples catalog by scanning example handlers."""
41 repo_dir = Path(__file__).resolve().parent.parent
42 source_root = repo_dir / "async-durable-execution-examples" / "src" / PACKAGE_PREFIX
43 examples = []
44 for path in sorted(source_root.rglob("*.py")):
45 if path.name in {"__init__.py", "__about__.py"}:
46 continue
48 example = build_example_entry(path, source_root)
49 if example is not None:
50 examples.append(example)
52 return {
53 "packageName": PACKAGE_NAME,
54 "examples": examples,
55 }
58def build_example_entry(path: Path, source_root: Path) -> dict[str, Any] | None:
59 """Build a catalog entry for a module if it exports a handler."""
60 tree = ast.parse(path.read_text(), filename=str(path))
61 handler_node = find_handler_node(tree)
62 if handler_node is None:
63 return None
65 relative_path = path.relative_to(source_root)
66 handler_module = ".".join([PACKAGE_PREFIX, *relative_path.with_suffix("").parts])
67 description = get_description(tree, handler_node, relative_path)
69 example: dict[str, Any] = {
70 "name": to_example_name(relative_path),
71 "description": description,
72 "handler": f"{handler_module}.handler",
73 "integration": True,
74 "durableConfig": DEFAULT_DURABLE_CONFIG.copy(),
75 "path": f"./src/{PACKAGE_PREFIX}/{relative_path.as_posix()}",
76 }
78 logging_config = SPECIAL_LOGGING_CONFIG.get(relative_path.as_posix())
79 if logging_config is not None:
80 example["loggingConfig"] = logging_config.copy()
82 return example
85def find_handler_node(
86 tree: ast.Module,
87) -> ast.FunctionDef | ast.AsyncFunctionDef | None:
88 """Return the top-level handler function node if present."""
89 for node in tree.body:
90 if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
91 if node.name == "handler":
92 return node
93 return None
96def get_description(
97 tree: ast.Module,
98 handler_node: ast.FunctionDef | ast.AsyncFunctionDef,
99 relative_path: Path,
100) -> str:
101 """Extract an example description from the handler or module docstring."""
102 handler_docstring = ast.get_docstring(handler_node)
103 if handler_docstring:
104 return first_line(handler_docstring)
106 module_docstring = ast.get_docstring(tree)
107 if module_docstring:
108 return first_line(module_docstring)
110 return f"Example for {to_example_name(relative_path)}."
113def first_line(docstring: str) -> str:
114 """Return the first non-empty line from a docstring."""
115 for line in docstring.splitlines():
116 stripped = line.strip()
117 if stripped:
118 return stripped
119 return ""
122def to_example_name(relative_path: Path) -> str:
123 """Convert a module path to a human-readable example name."""
124 parts = list(relative_path.with_suffix("").parts)
125 words: list[str] = []
126 previous_part_words: list[str] = []
127 for part in parts:
128 part_words = [word for word in part.split("_") if word]
129 if part_words[: len(previous_part_words)] == previous_part_words:
130 part_words = part_words[len(previous_part_words) :]
131 words.extend(part_words)
132 previous_part_words = [word for word in part.split("_") if word]
134 return " ".join(word.capitalize() for word in words if word)
137def load_catalog() -> dict[str, Any]:
138 """Generate the examples catalog from source."""
139 return build_examples_catalog()
142def build_template(
143 examples: list[dict[str, Any]],
144 *,
145 runtime: str = DEFAULT_RUNTIME,
146) -> dict[str, Any]:
147 """Build a SAM template for all examples."""
148 parameters: dict[str, Any] = {
149 "LambdaEndpoint": {
150 "Type": "String",
151 "Default": DEFAULT_LAMBDA_ENDPOINT,
152 },
153 "FunctionNamePrefix": {
154 "Type": "String",
155 "Default": "",
156 },
157 }
159 template: dict[str, Any] = {
160 "AWSTemplateFormatVersion": "2010-09-09",
161 "Transform": "AWS::Serverless-2016-10-31",
162 "Globals": {
163 "Function": {
164 "Runtime": runtime,
165 "Timeout": 60,
166 "MemorySize": 128,
167 "Environment": {
168 "Variables": {"AWS_ENDPOINT_URL_LAMBDA": {"Ref": "LambdaEndpoint"}}
169 },
170 }
171 },
172 "Parameters": parameters,
173 "Resources": {
174 SDK_LAYER_LOGICAL_ID: {
175 "Type": "AWS::Serverless::LayerVersion",
176 "Properties": {
177 "LayerName": {"Fn::Sub": "${FunctionNamePrefix}sdk"},
178 "Description": "async-durable-execution SDK for e2e functions",
179 "ContentUri": SDK_LAYER_CONTENT_URI,
180 "CompatibleRuntimes": [runtime],
181 "CompatibleArchitectures": ["x86_64"],
182 },
183 },
184 "DurableFunctionRole": {
185 "Type": "AWS::IAM::Role",
186 "Properties": {
187 "AssumeRolePolicyDocument": {
188 "Version": "2012-10-17",
189 "Statement": [
190 {
191 "Effect": "Allow",
192 "Principal": {"Service": "lambda.amazonaws.com"},
193 "Action": "sts:AssumeRole",
194 }
195 ],
196 },
197 "ManagedPolicyArns": [
198 "arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole"
199 ],
200 "Policies": [
201 {
202 "PolicyName": "DurableExecutionPolicy",
203 "PolicyDocument": {
204 "Version": "2012-10-17",
205 "Statement": [
206 {
207 "Effect": "Allow",
208 "Action": [
209 "lambda:CheckpointDurableExecution",
210 "lambda:GetDurableExecutionState",
211 "lambda:InvokeFunction",
212 ],
213 "Resource": "*",
214 }
215 ],
216 },
217 }
218 ],
219 },
220 },
221 },
222 }
224 for example in examples:
225 logical_id = to_logical_id(example["handler"])
226 function_name_suffix = to_function_name_suffix(example["handler"])
227 properties: dict[str, Any] = {
228 "CodeUri": "build/",
229 "Handler": example["handler"],
230 "Description": example["description"],
231 "Role": {"Fn::GetAtt": ["DurableFunctionRole", "Arn"]},
232 "Layers": [{"Ref": SDK_LAYER_LOGICAL_ID}],
233 "FunctionName": {
234 "Fn::Sub": f"${{FunctionNamePrefix}}{function_name_suffix}"
235 },
236 }
238 if "durableConfig" in example:
239 properties["DurableConfig"] = example["durableConfig"]
241 template["Resources"][logical_id] = {
242 "Type": "AWS::Serverless::Function",
243 "Properties": properties,
244 }
246 return template
249def validate_catalog_test_coverage(catalog: dict[str, Any]) -> None:
250 """Ensure every example test handler is represented in the examples catalog."""
251 catalog_handlers = {example["handler"] for example in catalog["examples"]}
252 repo_dir = Path(__file__).resolve().parent.parent
253 test_root = repo_dir / "async-durable-execution-examples" / "test_examples"
254 missing_handlers = sorted(
255 handler
256 for handler in load_test_handlers(test_root)
257 if handler not in catalog_handlers
258 )
259 if not missing_handlers:
260 return
262 missing_details = ", ".join(missing_handlers)
263 msg = (
264 "Example tests reference handlers missing from the generated examples catalog: "
265 f"{missing_details}"
266 )
267 raise SystemExit(msg)
270def generate_sam_template(
271 *,
272 output_path: Path | None = None,
273 runtime: str = DEFAULT_RUNTIME,
274) -> Path:
275 """Generate a SAM template for the full examples stack."""
276 catalog = load_catalog()
277 validate_catalog_test_coverage(catalog)
279 template = build_template(
280 catalog["examples"],
281 runtime=runtime,
282 )
284 repo_dir = Path(__file__).resolve().parent.parent
285 template_path = output_path or (
286 repo_dir / "async-durable-execution-examples" / "template.generated.json"
287 )
288 template_path.parent.mkdir(parents=True, exist_ok=True)
289 with template_path.open("w") as file:
290 json.dump(template, file, sort_keys=False, indent=2)
291 file.write("\n")
293 return template_path
296def main() -> int:
297 parser = argparse.ArgumentParser(description="Generate a SAM template for examples")
298 parser.add_argument(
299 "--output",
300 type=Path,
301 help="Write the generated template to this path",
302 )
303 parser.add_argument(
304 "--runtime",
305 default=DEFAULT_RUNTIME,
306 help=f"SAM Lambda runtime to use for generated functions (default: {DEFAULT_RUNTIME})",
307 )
308 args = parser.parse_args()
310 template_path = generate_sam_template(
311 output_path=args.output,
312 runtime=args.runtime,
313 )
314 print(f"Generated SAM template at {template_path}")
315 return 0
318if __name__ == "__main__":
319 raise SystemExit(main())