Add automation to creating boto patches

.github/workflows/maintenance-v1.yaml

@@ -25,7 +25,7 @@ jobs:
           cfn-lint --update-documentation
           scripts/update_specs_from_pricing.py
           scripts/update_serverless_aws_policies.py
-          scripts/update_schemas_from_boto.py
+          scripts/boto/update_schemas_from_boto.py
           scripts/update_schemas_from_aws_api.py
           cfn-lint --update-specs
           echo "date=$(date +'%Y-%m-%d')" >> $GITHUB_OUTPUT

scripts/boto/_automated_patches.py

@@ -0,0 +1,210 @@
+"""
+Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+SPDX-License-Identifier: MIT-0
+"""
+
+import json
+from pathlib import Path
+from typing import Any
+
+from _types import AllPatches, Patch, ResourcePatches
+
+skip = [
+    "account",
+    "chime",
+    "chimesdkidentity",
+    "chimesdkmessaging",
+    "chimesdkmeetings",
+    "chimesdkvoice",
+    "paymentcryptographydata",
+    "rdsdata",
+    "finspacedata",
+    "appconfigdata",
+    "iotjobsdata",
+    "dataexchange",
+    "bedrockruntime",
+    "swf",
+    "cloudhsm",
+    "cloudhsmv2",
+    "workdocs",
+]
+
+skip_property_names = ["State"]
+
+_fields = ["pattern", "enum"]
+
+
+def renamer(name):
+    manual_fixes = {
+        "acm": "CertificateManager",
+        "mq": "AmazonMQ",
+        "kafka": "MSK",
+        "firehose": "KinesisFirehose",
+        "es": "ElasticSearch",
+    }
+    if name in manual_fixes:
+        return manual_fixes[name].lower()
+
+    return name.replace("-", "").lower()
+
+
+def get_shapes(data: dict[str, Any], name: str):
+    shapes: dict[str, Any] = {}
+
+    input_shape = data.get("operations", {}).get(name, {}).get("input", {}).get("shape")
+    if not input_shape:
+        return shapes
+
+    for shape_name, shap_data in data.get("shapes", {}).items():
+        if "enum" in shap_data:
+            shapes[shape_name] = {"enum": shap_data.get("enum")}
+
+    return shapes
+
+
+def get_schema_create_operations(data: dict[str, Any]) -> list[str]:
+    results = []
+
+    action_prefixes = ["Put", "Add", "Create", "Register", "Allocate", "Start", "Run"]
+
+    for api in data.get("handlers", {}).get("create", {}).get("permissions", []):
+        if ":" not in api:
+            continue
+        api = api.split(":")[1]
+        for action_prefix in action_prefixes:
+            if api.startswith(action_prefix):
+                results.append(api)
+
+    return results
+
+
+def get_last_date(service_dir: Path) -> str:
+    last_date = "0000-00-00"
+    for date_dir in service_dir.iterdir():
+        if not date_dir.is_dir():
+            continue
+
+        if date_dir.name > last_date:
+            last_date = date_dir.name
+
+    return last_date
+
+
+def _per_resource_patch(
+    schema_data: dict[str, Any], boto_data: dict[str, Any], source: list[str]
+) -> ResourcePatches:
+    results: ResourcePatches = {}
+    create_operations = get_schema_create_operations(schema_data)
+    shapes = {}
+    for create_operation in create_operations:
+        shapes.update(get_shapes(boto_data, create_operation))
+        create_shape = (
+            boto_data.get("operations", {})
+            .get(create_operation, {})
+            .get("input", {})
+            .get("shape")
+        )
+
+        for member, member_data in (
+            boto_data.get("shapes", {}).get(create_shape, {}).get("members", {}).items()
+        ):
+            for p_name, p_data in schema_data.get("properties", {}).items():
+                if p_name in skip_property_names:
+                    continue
+                if p_name.lower() == member.lower():
+
+                    path = f"/properties/{p_name}"
+
+                    if "$ref" in p_data:
+                        pointer = p_data["$ref"].split("/")
+                        p_data = schema_data.get(pointer[1], {}).get(pointer[2], {})
+                        if not p_data:
+                            continue
+                        path = f"/{'/'.join(pointer[1:])}"
+
+                    # skip if we already have an enum or pattern
+                    if any([p_data.get(field) for field in _fields]):
+                        continue
+
+                    member_shape_name = member_data.get("shape")
+                    member_shape = boto_data.get("shapes", {}).get(
+                        member_shape_name, {}
+                    )
+
+                    if not any([member_shape.get(field) for field in _fields]):
+                        continue
+
+                    results[path] = Patch(
+                        source=source,
+                        shape=member_shape_name,
+                    )
+
+    return results
+
+
+def get_resource_patches(
+    service_dir: Path, schema_path: Path, service_name: str, last_date: str
+) -> AllPatches:
+
+    results: AllPatches = {}
+
+    services_file = Path(f"{service_dir}/{last_date}/service-2.json")
+    if not services_file.exists():
+        return results
+
+    boto_data = {}
+    with open(services_file, "r") as f:
+        boto_data = json.load(f)
+
+    if not boto_data:
+        return results
+
+    resources = list(schema_path.glob(f"aws-{service_name}-*.json"))
+    if not resources:
+        print(f"No resource files found for {service_name}")
+
+    for resource in resources:
+        with open(resource, "r") as f:
+            schema_data = json.load(f)
+
+            resource_type = schema_data.get("typeName", "")
+            if resource_type not in results:
+                results[resource_type] = {}
+
+            results[resource_type].update(
+                _per_resource_patch(
+                    schema_data, boto_data, [service_dir.name, last_date]
+                )
+            )
+
+    return results
+
+
+def each_boto_service(boto_path: Path, schema_path: Path) -> AllPatches:
+    results: AllPatches = {}
+    _results: AllPatches = {}
+    boto_path = boto_path / "botocore-master" / "botocore" / "data"
+
+    for service_dir in boto_path.iterdir():
+        if not service_dir.is_dir():
+            continue
+
+        service_name = renamer(service_dir.name)
+
+        if service_name in skip:
+            continue
+
+        last_date = get_last_date(service_dir)
+
+        _results = get_resource_patches(
+            service_dir, schema_path, service_name, last_date
+        )
+        for type_name, patches in _results.items():
+            if patches:
+                results[type_name] = patches
+
+    return results
+
+
+def build_automated_patches(boto_path: Path, schema_path: Path) -> AllPatches:
+    return each_boto_service(boto_path, schema_path)