Automate the Cloudformation resources table generation (#243)

sannya-singal · remotesynth · k-a-il · web-flow · commit 75ac1bd236a1 · 2025-10-23T14:09:58.000-04:00
Co-authored-by: Brian Rinaldi &lt;brian.rinaldi@gmail.com&gt;
Co-authored-by: Anastasia Dusak &lt;61540676+k-a-il@users.noreply.github.com&gt;
diff --git a/.github/workflows/docs-parity-updates.yml b/.github/workflows/docs-parity-updates.yml
@@ -103,6 +103,17 @@ jobs:
           WORKFLOW: "AWS / Build, Test, Push"
           RESOURCE_FOLDER: "metrics-implementation-details/community"
 
+      - name: Download CloudFormation catalog from Pro pipeline (GitHub)
+        working-directory: docs
+        continue-on-error: true
+        run: /tmp/get_latest_github_metrics.sh ./target main
+        env:
+          GITHUB_TOKEN: ${{ secrets.PRO_ACCESS_TOKEN }}
+          REPOSITORY_NAME: localstack-pro
+          ARTIFACT_ID: iac-catalog-assets
+          WORKFLOW: "IaC – Generate Catalog Assets"
+          RESOURCE_FOLDER: "iac-catalog-assets"
+
       - name: Download coverage data for K8s pipeline (acceptance tests) (GitHub)
         working-directory: docs
         run: /tmp/get_latest_github_metrics.sh ./target main
@@ -139,6 +150,12 @@ jobs:
           python3 -m scripts.create_data_coverage -i target/metrics-implementation-details -r target/metrics-raw -o target/updated_coverage -s src/data/coverage/service_display_name.json
           mv -f target/updated_coverage/data/*.json src/data/coverage
      
+      - name: Generate CloudFormation Coverage Tables
+        working-directory: docs
+        continue-on-error: true
+        run: |
+            python3 scripts/create_cfn_coverage_table.py --cfn-json target/iac-catalog-assets/cfn_resources.json
+
       - name: Check for changes
         id: check-for-changes
         working-directory: docs
@@ -147,7 +164,8 @@ jobs:
           # Check against the PR branch if it exists, otherwise against the main
           # Store the result in resources/diff-check.log and store the diff count in the GitHub Action output "diff-count"
           mkdir -p resources
-          (git diff --name-only origin/parity-coverage-auto-updates data/coverage/ 2>/dev/null || git diff --name-only origin/${{ github.event.inputs.targetBranch || 'main' }} src/data/coverage/ 2>/dev/null) | tee -a resources/diff-check.log
+          FILES_TO_CHECK="src/data/coverage/ src/content/docs/aws/services/cloudformation.mdx"
+          (git diff --name-only origin/parity-coverage-auto-updates $FILES_TO_CHECK 2>/dev/null || git diff --name-only origin/${{ github.event.inputs.targetBranch || 'main' }} $FILES_TO_CHECK 2>/dev/null) | tee -a resources/diff-check.log
           echo "diff-count=$(cat resources/diff-check.log | wc -l)" >> $GITHUB_OUTPUT
 
       - name: Read PR markdown template
diff --git a/scripts/create_cfn_coverage_table.py b/scripts/create_cfn_coverage_table.py
@@ -0,0 +1,252 @@
+from __future__ import annotations
+
+import argparse
+import json
+import re
+from io import StringIO
+from pathlib import Path
+from typing import IO, Callable
+
+from dataclasses import dataclass
+
+COMMUNITY_HEADING_PATTERN = r"####\s*Community image"
+PRO_HEADING_PATTERN = r"\n####\s*Pro image"
+API_COVERAGE_HEADING_PATTERN = r"^\s*##\s|$\Z"
+DEFAULT_PAGE_PATH = "src/content/docs/aws/services/cloudformation.mdx"
+
+
+@dataclass
+class ColumnConfig:
+    header: str
+    key: str
+    alignment: str = "left"
+    formatter: Callable[[any], str] | None = None
+
+    def format_value(self, value: any) -> str:
+        if self.formatter:
+            return self.formatter(value)
+        return str(value) if value is not None else "-"
+
+
+def bool_formatter(value: bool):
+    return "✅" if value else "-"
+
+
+@dataclass
+class TableConfig:
+    columns: list[ColumnConfig]
+    sort_by: str | None = None
+
+    def get_headers(self) -> list[str]:
+        return [col.header for col in self.columns]
+
+
+class MarkdownTableGenerator:
+    def __init__(self, config: TableConfig):
+        self.config = config
+
+    def _calculate_column_widths(self, data: list[dict[str, any]]) -> list[int]:
+        widths = []
+
+        for col in self.config.columns:
+            max_width = len(col.header)
+
+            for row in data:
+                value = row.get(col.key, "")
+                formatted_value = col.format_value(value)
+                max_width = max(max_width, len(formatted_value))
+
+            widths.append(max_width)
+
+        return widths
+
+    def _get_alignment_separator(self, alignment: str, width: int) -> str:
+        if alignment == "right":
+            return f"{'-' * (width + 1)}:"
+        elif alignment == "center":
+            return f":{'-' * width}:"
+        else:
+            return f"{'-' * width}"
+
+    def _format_cell(self, value: str, width: int, alignment: str) -> str:
+        if alignment == "right":
+            return value.rjust(width)
+        elif alignment == "center":
+            return value.center(width)
+        else:
+            return value.ljust(width)
+
+    def _write_header_row(self, writer: IO[str], widths: list[int]) -> None:
+        headers = self.config.get_headers()
+        formatted_headers = []
+
+        for i, header in enumerate(headers):
+            alignment = self.config.columns[i].alignment
+            formatted_header = self._format_cell(header, widths[i], alignment)
+            formatted_headers.append(formatted_header)
+
+        writer.write(f"| {' | '.join(formatted_headers)} |\n")
+
+    def _write_separator_row(self, writer: IO[str], widths: list[int]) -> None:
+        separators = []
+        for i, col in enumerate(self.config.columns):
+            separator = self._get_alignment_separator(col.alignment, widths[i])
+            separators.append(separator)
+
+        writer.write(f"|{'|'.join(separators)}|\n")
+
+    def _write_data_rows(
+        self, writer: IO[str], data: list[Dict[str, Any]], widths: list[int]
+    ) -> None:
+        if self.config.sort_by:
+            data = sorted(data, key=lambda x: x.get(self.config.sort_by, ""))
+
+        for row in data:
+            formatted_cells = []
+
+            for i, col in enumerate(self.config.columns):
+                value = row.get(col.key, "")
+                formatted_value = col.format_value(value)
+                formatted_cell = self._format_cell(
+                    formatted_value, widths[i], col.alignment
+                )
+                formatted_cells.append(formatted_cell)
+
+            writer.write(f"| {' | '.join(formatted_cells)} |\n")
+
+    def generate_table(self, data: list[dict[str, any]] | None) -> str:
+        if not data:
+            return ""
+        
+        buffer = StringIO()
+        widths = self._calculate_column_widths(data)
+
+        self._write_header_row(buffer, widths)
+        self._write_separator_row(buffer, widths)
+        self._write_data_rows(buffer, data, widths)
+
+        table = buffer.getvalue().rstrip("\n") + "\n"
+        return table
+
+class CloudFormationDataTransformer():
+    def transform(self, section_data: dict[str, any] | None) -> list[dict[str, any]]:
+        if not section_data:
+            return []
+
+        rows = []
+
+        for resource_type, metadata in section_data.items():
+            methods = set(metadata.get("methods", []))
+
+            row = {
+                "resource": resource_type,
+                "create": "Create" in methods,
+                "delete": "Delete" in methods,
+                "update": "Update" in methods,
+            }
+            rows.append(row)
+        return rows
+
+
+def create_argument_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(
+        description="Update CloudFormation Resources tables in docs"
+    )
+    parser.add_argument(
+        "--cfn-json",
+        required=True,
+        type=Path,
+        help="Path to iac-catalog-assets/cfn_resources.json in downloaded artifacts",
+    )
+    parser.add_argument(
+        "--md-file",
+        required=False,
+        type=Path,
+        default=str(DEFAULT_PAGE_PATH),
+        help="Markdown file which needs to be updated",
+    )
+
+    return parser
+
+
+def _load_cfn_file(cfn_file_path: Path) -> dict[str, any]:
+    try:
+        with cfn_file_path.open("r", encoding="utf-8") as f:
+            return json.load(f)
+    except json.JSONDecodeError as e:
+        raise ValueError(f"Invalid JSON in cfn json file: {e}")
+
+
+def replace_content_between(
+    content: str,
+    starting_rx: str,
+    ending_rx: str,
+    replacement_block: str,
+) -> str:
+    # Build a regex that replaces the content between two headings starting_rx and ending_rx.
+    # Group1 - start heading
+    # Group2 - content; lookahead preserves end boundary.
+    pattern = re.compile(
+        rf"(^{starting_rx}\s*\n)(.*?)(?={ending_rx})",
+        re.DOTALL | re.MULTILINE,
+    )
+
+    match = pattern.search(content)
+    if not match:
+        raise ValueError(
+            f"Could not find section with heading pattern: {starting_rx!r}"
+        )
+
+    heading = match.group(1)
+    replacement = f"{heading}{replacement_block}" if replacement_block else heading
+
+    return pattern.sub(replacement, content, count=1)
+
+
+def main():
+    parser = create_argument_parser()
+    args = parser.parse_args()
+
+    table_config = TableConfig(
+        columns=[
+            ColumnConfig("Resource", "resource", "left"),
+            ColumnConfig("Create", "create", "right", bool_formatter),
+            ColumnConfig("Delete", "delete", "right", bool_formatter),
+            ColumnConfig("Update", "update", "right", bool_formatter),
+        ]
+    )
+
+    table_generator = MarkdownTableGenerator(table_config)
+    data_transformer = CloudFormationDataTransformer()
+
+    cfn_catalog = _load_cfn_file(args.cfn_json)
+
+    community_data = data_transformer.transform(cfn_catalog.get("community"))
+    pro_data = data_transformer.transform(cfn_catalog.get("pro"))
+
+    community_table = table_generator.generate_table(community_data)
+    pro_table = table_generator.generate_table(pro_data)
+
+    original_doc = args.md_file.read_text(encoding="utf-8")
+    updated_doc = original_doc
+
+    updated_doc = replace_content_between(
+        content=original_doc,
+        starting_rx=COMMUNITY_HEADING_PATTERN,
+        ending_rx=PRO_HEADING_PATTERN,
+        replacement_block=community_table,
+    )
+
+    updated_doc = replace_content_between(
+        content=updated_doc,
+        starting_rx=PRO_HEADING_PATTERN,
+        ending_rx=API_COVERAGE_HEADING_PATTERN,
+        replacement_block=pro_table,
+    )
+
+    if updated_doc != original_doc:
+        args.md_file.write_text(updated_doc)
+
+
+if __name__ == "__main__":
+    main()
