Spaces:

Vishvam10
/

cloudenv

Running

cloudenv / scripts /generate_explicit_aws_services.py

Vishvam10

initial space commit

b404de3 7 months ago

28.5 kB

# Generate explicit BaseService subclasses under :
# server/providers/aws/*_service.py from AWS_SERVICE_OPERATIONS


from __future__ import annotations

import sys
from pathlib import Path
from textwrap import dedent

from cloudenv.server.providers.aws.base import pascal_to_snake
from cloudenv.server.providers.aws.operations_data import (
    AWS_SERVICE_OPERATIONS as OPS,
)

ROOT = Path(__file__).resolve().parents[1]
PROVIDER = ROOT / "server" / "providers" / "aws"

sys.path.insert(0, str(ROOT))
sys.path.insert(0, str(Path(__file__).resolve().parent))

# boto client id -> (module stem without _service, Python class name)
SERVICE_SPECS: dict[str, tuple[str, str]] = {
    "secretsmanager": ("secrets_manager_service", "SecretsManagerService"),
    "ssm": ("ssm_service", "SSMService"),
    "logs": ("cloudwatch_logs_service", "CloudWatchLogsService"),
    "events": ("eventbridge_service", "EventBridgeService"),
    "kinesis": ("kinesis_service", "KinesisService"),
    "cloudwatch": ("cloudwatch_service", "CloudWatchService"),
    "stepfunctions": ("stepfunctions_service", "StepFunctionsService"),
    "firehose": ("firehose_service", "FirehoseService"),
    "ecs": ("ecs_service", "ECSService"),
    "glue": ("glue_service", "GlueService"),
    "athena": ("athena_service", "AthenaService"),
    "ses": ("ses_service", "SESService"),
    "sesv2": ("sesv2_service", "SESv2Service"),
    "acm": ("acm_service", "ACMService"),
    "wafv2": ("wafv2_service", "WAFv2Service"),
    "route53": ("route53_service", "Route53Service"),
    "rds": ("rds_service", "RDSService"),
    "elasticache": ("elasticache_service", "ElastiCacheService"),
    "sqs": ("sqs_service", "SQSService"),
    "dynamodb": ("dynamodb_service", "DynamoDBService"),
    "lambda": ("lambda_service", "LambdaService"),
    "iam": ("iam_service", "IAMService"),
    "s3": ("s3_service", "S3Service"),
    "ec2": ("ec2_service", "EC2Service"),
    "apigateway": ("apigateway_service", "APIGatewayService"),
    "apigatewayv2": ("apigatewayv2_service", "APIGatewayV2Service"),
    "cognito-idp": ("cognito_idp_service", "CognitoIDPService"),
    "cognito-identity": ("cognito_identity_service", "CognitoIdentityService"),
}

# (boto_id, OperationName) -> boto method name when it differs from pascal_to_snake(OperationName)
METHOD_OVERRIDES: dict[tuple[str, str], str] = {
    # pascal_to_snake turns trailing ACLs into ac_ls; boto uses acls
    ("wafv2", "ListWebACLs"): "list_web_acls",
}


def boto_method_name(boto_id: str, operation: str) -> str:
    return METHOD_OVERRIDES.get(
        (boto_id, operation), pascal_to_snake(operation)
    )


def render_methods(boto_id: str, operations: list[str]) -> str:
    lines: list[str] = []
    for op in operations:
        snake = boto_method_name(boto_id, op)
        lines.append(
            f"    def {op}(self, **kwargs):\n"
            f'        """Forward to boto3 {snake} (IAM: {boto_id}:{op})."""\n'
            f"        return self.client.{snake}(**kwargs)\n"
        )
    return "\n".join(lines)


def render_lifecycle(boto_id: str, class_name: str) -> str:
    """reset / health / describe_state blocks."""
    if boto_id == "s3":
        return dedent(
            """
    def reset(self) -> None:
        try:
            buckets = self.client.list_buckets().get("Buckets", [])
            for b in buckets:
                name = b["Name"]
                objs = self.client.list_objects_v2(Bucket=name).get("Contents", [])
                for obj in objs:
                    self.client.delete_object(Bucket=name, Key=obj["Key"])
                self.client.delete_bucket(Bucket=name)
        except Exception:
            pass
    def health(self) -> float:
        try:
            self.client.list_buckets()
            return 1.0
        except Exception:
            return 0.0
    def describe_state(self) -> dict:
        try:
            buckets = self.client.list_buckets().get("Buckets", [])
            result = []
            for b in buckets:
                name = b["Name"]
                objs = self.client.list_objects_v2(Bucket=name).get("Contents", [])
                result.append({"name": name, "object_count": len(objs)})
            return {"buckets": result}
        except Exception:
            return {"buckets": []}
"""
        )

    if boto_id == "sqs":
        return dedent(
            """
    def reset(self) -> None:
        try:
            queues = self.client.list_queues().get("QueueUrls", [])
            for q in queues:
                self.client.delete_queue(QueueUrl=q)
        except Exception:
            pass
    def health(self) -> float:
        try:
            self.client.list_queues()
            return 1.0
        except Exception:
            return 0.0
    def describe_state(self) -> dict:
        try:
            urls = self.client.list_queues().get("QueueUrls", [])
            queues = []
            for q in urls:
                attrs = self.client.get_queue_attributes(
                    QueueUrl=q, AttributeNames=["ApproximateNumberOfMessages"]
                )["Attributes"]
                queues.append(
                    {"url": q, "messages": int(attrs.get("ApproximateNumberOfMessages", 0))}
                )
            return {"queues": queues}
        except Exception:
            return {"queues": []}
"""
        )

    if boto_id == "dynamodb":
        return dedent(
            """
    def reset(self) -> None:
        try:
            tables = self.client.list_tables().get("TableNames", [])
            for t in tables:
                self.client.delete_table(TableName=t)
        except Exception:
            pass
    def health(self) -> float:
        try:
            self.client.list_tables()
            return 1.0
        except Exception:
            return 0.0
    def describe_state(self) -> dict:
        try:
            tables = self.client.list_tables().get("TableNames", [])
            return {"tables": tables}
        except Exception:
            return {"tables": []}
"""
        )

    if boto_id == "lambda":
        return dedent(
            """
    def reset(self) -> None:
        try:
            funcs = self.client.list_functions().get("Functions", [])
            for f in funcs:
                self.client.delete_function(FunctionName=f["FunctionName"])
        except Exception:
            pass
    def health(self) -> float:
        try:
            self.client.list_functions()
            return 1.0
        except Exception:
            return 0.0
    def describe_state(self) -> dict:
        try:
            funcs = self.client.list_functions().get("Functions", [])
            return {"functions": [f["FunctionName"] for f in funcs]}
        except Exception:
            return {"functions": []}
"""
        )

    if boto_id == "iam":
        return dedent(
            """
    def reset(self) -> None:
        try:
            policies = self.client.list_policies(Scope="Local").get("Policies", [])
            for p in policies:
                try:
                    self.client.delete_policy(PolicyArn=p["Arn"])
                except Exception:
                    pass
            roles = self.client.list_roles().get("Roles", [])
            for r in roles:
                try:
                    self.client.delete_role(RoleName=r["RoleName"])
                except Exception:
                    pass
        except Exception:
            pass
    def health(self) -> float:
        try:
            self.client.list_roles()
            return 1.0
        except Exception:
            return 0.0
    def describe_state(self) -> dict:
        try:
            roles = self.client.list_roles().get("Roles", [])
            return {"roles": [r["RoleName"] for r in roles]}
        except Exception:
            return {"roles": []}
"""
        )

    if boto_id == "secretsmanager":
        return dedent(
            """
    def reset(self) -> None:
        try:
            for s in self.client.list_secrets().get("SecretList", []):
                try:
                    self.client.delete_secret(
                        SecretId=s["ARN"], ForceDeleteWithoutRecovery=True
                    )
                except Exception:
                    pass
        except Exception:
            pass
    def health(self) -> float:
        try:
            self.client.list_secrets(MaxResults=1)
            return 1.0
        except Exception:
            return 0.0
    def describe_state(self) -> dict:
        try:
            n = len(self.client.list_secrets().get("SecretList", []))
            return {"secrets": n}
        except Exception:
            return {"secrets": 0}
"""
        )

    if boto_id == "ssm":
        return dedent(
            """
    def reset(self) -> None:
        try:
            paginator = self.client.get_paginator("describe_parameters")
            names = []
            for page in paginator.paginate():
                for p in page.get("Parameters", []):
                    names.append(p["Name"])
            for name in names:
                try:
                    self.client.delete_parameter(Name=name)
                except Exception:
                    pass
        except Exception:
            pass
    def health(self) -> float:
        try:
            self.client.describe_parameters(MaxResults=1)
            return 1.0
        except Exception:
            return 0.0
    def describe_state(self) -> dict:
        try:
            r = self.client.describe_parameters(MaxResults=50)
            return {"parameters": len(r.get("Parameters", []))}
        except Exception:
            return {"parameters": 0}
"""
        )

    if boto_id == "logs":
        return dedent(
            """
    def reset(self) -> None:
        try:
            groups = self.client.describe_log_groups().get("logGroups", [])
            for g in groups:
                try:
                    self.client.delete_log_group(logGroupName=g["logGroupName"])
                except Exception:
                    pass
        except Exception:
            pass
    def health(self) -> float:
        try:
            self.client.describe_log_groups(limit=1)
            return 1.0
        except Exception:
            return 0.0
    def describe_state(self) -> dict:
        try:
            groups = self.client.describe_log_groups().get("logGroups", [])
            return {"log_groups": len(groups)}
        except Exception:
            return {"log_groups": 0}
"""
        )

    if boto_id == "events":
        return dedent(
            """
    def reset(self) -> None:
        try:
            buses = self.client.list_event_buses().get("EventBuses", [])
            for b in buses:
                name = b["Name"]
                if name == "default":
                    continue
                try:
                    self.client.delete_event_bus(Name=name)
                except Exception:
                    pass
        except Exception:
            pass
    def health(self) -> float:
        try:
            self.client.list_event_buses(Limit=1)
            return 1.0
        except Exception:
            return 0.0
    def describe_state(self) -> dict:
        try:
            buses = self.client.list_event_buses().get("EventBuses", [])
            return {"event_buses": len(buses)}
        except Exception:
            return {"event_buses": 0}
"""
        )

    if boto_id == "kinesis":
        return dedent(
            """
    def reset(self) -> None:
        try:
            names = self.client.list_streams().get("StreamNames", [])
            for n in names:
                try:
                    self.client.delete_stream(StreamName=n, EnforceConsumerDeletion=True)
                except Exception:
                    pass
        except Exception:
            pass
    def health(self) -> float:
        try:
            self.client.list_streams(Limit=1)
            return 1.0
        except Exception:
            return 0.0
    def describe_state(self) -> dict:
        try:
            names = self.client.list_streams().get("StreamNames", [])
            return {"streams": len(names)}
        except Exception:
            return {"streams": 0}
"""
        )

    if boto_id == "cloudwatch":
        return dedent(
            """
    def reset(self) -> None:
        try:
            alarms = self.client.describe_alarms().get("MetricAlarms", [])
            names = [a["AlarmName"] for a in alarms]
            if names:
                self.client.delete_alarms(AlarmNames=names)
            boards = self.client.list_dashboards().get("DashboardEntries", [])
            for d in boards:
                try:
                    self.client.delete_dashboards(DashboardNames=[d["DashboardName"]])
                except Exception:
                    pass
        except Exception:
            pass
    def health(self) -> float:
        try:
            self.client.list_metrics(Namespace="AWS/EC2")
            return 1.0
        except Exception:
            return 0.0
    def describe_state(self) -> dict:
        try:
            alarms = self.client.describe_alarms().get("MetricAlarms", [])
            return {"alarms": len(alarms)}
        except Exception:
            return {"alarms": 0}
"""
        )

    if boto_id == "stepfunctions":
        return dedent(
            """
    def reset(self) -> None:
        try:
            sms = self.client.list_state_machines().get("stateMachines", [])
            for sm in sms:
                try:
                    self.client.delete_state_machine(stateMachineArn=sm["stateMachineArn"])
                except Exception:
                    pass
        except Exception:
            pass
    def health(self) -> float:
        try:
            self.client.list_state_machines(maxResults=1)
            return 1.0
        except Exception:
            return 0.0
    def describe_state(self) -> dict:
        try:
            sms = self.client.list_state_machines().get("stateMachines", [])
            return {"state_machines": len(sms)}
        except Exception:
            return {"state_machines": 0}
"""
        )

    if boto_id == "firehose":
        return dedent(
            """
    def reset(self) -> None:
        try:
            names = self.client.list_delivery_streams().get("DeliveryStreamNames", [])
            for n in names:
                try:
                    self.client.delete_delivery_stream(DeliveryStreamName=n)
                except Exception:
                    pass
        except Exception:
            pass
    def health(self) -> float:
        try:
            self.client.list_delivery_streams(Limit=1)
            return 1.0
        except Exception:
            return 0.0
    def describe_state(self) -> dict:
        try:
            names = self.client.list_delivery_streams().get("DeliveryStreamNames", [])
            return {"delivery_streams": len(names)}
        except Exception:
            return {"delivery_streams": 0}
"""
        )

    if boto_id == "ecs":
        return dedent(
            """
    def reset(self) -> None:
        try:
            clusters = self.client.list_clusters().get("clusterArns", [])
            for arn in clusters:
                try:
                    self.client.delete_cluster(cluster=arn)
                except Exception:
                    pass
        except Exception:
            pass
    def health(self) -> float:
        try:
            self.client.list_clusters(maxResults=1)
            return 1.0
        except Exception:
            return 0.0
    def describe_state(self) -> dict:
        try:
            clusters = self.client.list_clusters().get("clusterArns", [])
            return {"clusters": len(clusters)}
        except Exception:
            return {"clusters": 0}
"""
        )

    if boto_id == "glue":
        return dedent(
            """
    def reset(self) -> None:
        try:
            dbs = self.client.get_databases().get("DatabaseList", [])
            for db in dbs:
                name = db["Name"]
                if name == "default":
                    continue
                try:
                    self.client.delete_database(Name=name)
                except Exception:
                    pass
        except Exception:
            pass
    def health(self) -> float:
        try:
            self.client.get_databases(MaxResults=1)
            return 1.0
        except Exception:
            return 0.0
    def describe_state(self) -> dict:
        try:
            dbs = self.client.get_databases().get("DatabaseList", [])
            return {"databases": len(dbs)}
        except Exception:
            return {"databases": 0}
"""
        )

    if boto_id == "athena":
        return dedent(
            """
    def reset(self) -> None:
        try:
            wgs = self.client.list_work_groups().get("WorkGroups", [])
            for wg in wgs:
                name = wg["Name"]
                if name == "primary":
                    continue
                try:
                    self.client.delete_work_group(WorkGroup=name, RecursiveDeleteOption=True)
                except Exception:
                    pass
        except Exception:
            pass
    def health(self) -> float:
        try:
            self.client.list_work_groups(MaxResults=1)
            return 1.0
        except Exception:
            return 0.0
    def describe_state(self) -> dict:
        try:
            wgs = self.client.list_work_groups().get("WorkGroups", [])
            return {"work_groups": len(wgs)}
        except Exception:
            return {"work_groups": 0}
"""
        )

    if boto_id in ("ses", "sesv2"):
        return dedent(
            """
    def reset(self) -> None:
        pass
    def health(self) -> float:
        try:
            if hasattr(self.client, "list_identities"):
                self.client.list_identities(MaxItems=1)
            else:
                self.client.list_email_identities(MaxResults=1)
            return 1.0
        except Exception:
            return 0.0
    def describe_state(self) -> dict:
        return {}
"""
        )

    if boto_id == "acm":
        return dedent(
            """
    def reset(self) -> None:
        try:
            certs = self.client.list_certificates().get("CertificateSummaryList", [])
            for c in certs:
                try:
                    self.client.delete_certificate(CertificateArn=c["CertificateArn"])
                except Exception:
                    pass
        except Exception:
            pass
    def health(self) -> float:
        try:
            self.client.list_certificates(MaxItems=1)
            return 1.0
        except Exception:
            return 0.0
    def describe_state(self) -> dict:
        try:
            certs = self.client.list_certificates().get("CertificateSummaryList", [])
            return {"certificates": len(certs)}
        except Exception:
            return {"certificates": 0}
"""
        )

    if boto_id == "wafv2":
        return dedent(
            """
    def reset(self) -> None:
        pass
    def health(self) -> float:
        try:
            self.client.list_web_acls(Scope="REGIONAL", Limit=1)
            return 1.0
        except Exception:
            return 0.0
    def describe_state(self) -> dict:
        return {}
"""
        )

    if boto_id == "route53":
        return dedent(
            """
    def reset(self) -> None:
        try:
            zones = self.client.list_hosted_zones().get("HostedZones", [])
            for z in zones:
                try:
                    self.client.delete_hosted_zone(Id=z["Id"])
                except Exception:
                    pass
        except Exception:
            pass
    def health(self) -> float:
        try:
            self.client.list_hosted_zones(MaxItems="1")
            return 1.0
        except Exception:
            return 0.0
    def describe_state(self) -> dict:
        try:
            zones = self.client.list_hosted_zones().get("HostedZones", [])
            return {"hosted_zones": len(zones)}
        except Exception:
            return {"hosted_zones": 0}
"""
        )

    if boto_id == "rds":
        return dedent(
            """
    def reset(self) -> None:
        try:
            dbs = self.client.describe_db_instances().get("DBInstances", [])
            for db in dbs:
                ident = db["DBInstanceIdentifier"]
                try:
                    self.client.delete_db_instance(
                        DBInstanceIdentifier=ident, SkipFinalSnapshot=True
                    )
                except Exception:
                    pass
        except Exception:
            pass
    def health(self) -> float:
        try:
            self.client.describe_db_instances(MaxRecords=20)
            return 1.0
        except Exception:
            return 0.0
    def describe_state(self) -> dict:
        try:
            dbs = self.client.describe_db_instances().get("DBInstances", [])
            return {"db_instances": len(dbs)}
        except Exception:
            return {"db_instances": 0}
"""
        )

    if boto_id == "elasticache":
        return dedent(
            """
    def reset(self) -> None:
        try:
            clusters = self.client.describe_cache_clusters().get("CacheClusters", [])
            for c in clusters:
                cid = c["CacheClusterId"]
                try:
                    self.client.delete_cache_cluster(CacheClusterId=cid)
                except Exception:
                    pass
        except Exception:
            pass
    def health(self) -> float:
        try:
            self.client.describe_cache_clusters(MaxRecords=20)
            return 1.0
        except Exception:
            return 0.0
    def describe_state(self) -> dict:
        try:
            clusters = self.client.describe_cache_clusters().get("CacheClusters", [])
            return {"cache_clusters": len(clusters)}
        except Exception:
            return {"cache_clusters": 0}
"""
        )

    if boto_id == "ec2":
        return dedent(
            """
    def reset(self) -> None:
        try:
            res = self.client.describe_instances()
            for r in res.get("Reservations", []):
                for inst in r.get("Instances", []):
                    iid = inst["InstanceId"]
                    try:
                        self.client.terminate_instances(InstanceIds=[iid])
                    except Exception:
                        pass
        except Exception:
            pass
    def health(self) -> float:
        try:
            self.client.describe_regions()
            return 1.0
        except Exception:
            return 0.0
    def describe_state(self) -> dict:
        try:
            res = self.client.describe_instances()
            n = sum(len(r.get("Instances", [])) for r in res.get("Reservations", []))
            return {"instances": n}
        except Exception:
            return {"instances": 0}
"""
        )

    if boto_id == "apigateway":
        return dedent(
            """
    def reset(self) -> None:
        try:
            apis = self.client.get_rest_apis().get("items", [])
            for a in apis:
                try:
                    self.client.delete_rest_api(restApiId=a["id"])
                except Exception:
                    pass
        except Exception:
            pass
    def health(self) -> float:
        try:
            self.client.get_rest_apis(limit=1)
            return 1.0
        except Exception:
            return 0.0
    def describe_state(self) -> dict:
        try:
            apis = self.client.get_rest_apis().get("items", [])
            return {"rest_apis": len(apis)}
        except Exception:
            return {"rest_apis": 0}
"""
        )

    if boto_id == "apigatewayv2":
        return dedent(
            """
    def reset(self) -> None:
        try:
            apis = self.client.get_apis().get("Items", [])
            for a in apis:
                try:
                    self.client.delete_api(ApiId=a["ApiId"])
                except Exception:
                    pass
        except Exception:
            pass
    def health(self) -> float:
        try:
            self.client.get_apis(MaxResults="1")
            return 1.0
        except Exception:
            return 0.0
    def describe_state(self) -> dict:
        try:
            apis = self.client.get_apis().get("Items", [])
            return {"apis": len(apis)}
        except Exception:
            return {"apis": 0}
"""
        )

    if boto_id == "cognito-idp":
        return dedent(
            """
    def reset(self) -> None:
        try:
            pools = self.client.list_user_pools(MaxResults=60).get("UserPools", [])
            for p in pools:
                try:
                    self.client.delete_user_pool(UserPoolId=p["Id"])
                except Exception:
                    pass
        except Exception:
            pass
    def health(self) -> float:
        try:
            self.client.list_user_pools(MaxResults=1)
            return 1.0
        except Exception:
            return 0.0
    def describe_state(self) -> dict:
        try:
            pools = self.client.list_user_pools(MaxResults=60).get("UserPools", [])
            return {"user_pools": len(pools)}
        except Exception:
            return {"user_pools": 0}
"""
        )

    if boto_id == "cognito-identity":
        return dedent(
            """
    def reset(self) -> None:
        try:
            pools = self.client.list_identity_pools(MaxResults=60).get("IdentityPools", [])
            for p in pools:
                try:
                    self.client.delete_identity_pool(IdentityPoolId=p["IdentityPoolId"])
                except Exception:
                    pass
        except Exception:
            pass
    def health(self) -> float:
        try:
            self.client.list_identity_pools(MaxResults=1)
            return 1.0
        except Exception:
            return 0.0
    def describe_state(self) -> dict:
        try:
            pools = self.client.list_identity_pools(MaxResults=60).get("IdentityPools", [])
            return {"identity_pools": len(pools)}
        except Exception:
            return {"identity_pools": 0}
"""
        )

    # default
    return dedent(
        """
    def reset(self) -> None:
        pass
    def health(self) -> float:
        return 1.0
    def describe_state(self) -> dict:
        return {}
"""
    )


def render_file(
    boto_id: str, module_stem: str, class_name: str, operations: list[str]
) -> str:
    methods = render_methods(boto_id, operations)
    life = render_lifecycle(boto_id, class_name)
    doc = f'"""Explicit {class_name} — boto `{boto_id}` (IAM: `{boto_id}:<Operation>`)."""'
    return dedent(
        f'''\
{doc}
from __future__ import annotations
from botocore.client import BaseClient
from cloudenv.server.providers.aws.client import get_client
from cloudenv.server.providers.aws.base import BaseService
class {class_name}(BaseService):
    def __init__(self) -> None:
        self.client: BaseClient = get_client("{boto_id}")
{methods}
{life}
'''
    )


def validate_against_boto() -> list[str]:
    import boto3

    errors: list[str] = []
    for boto_id, ops in OPS.items():
        if boto_id not in SERVICE_SPECS:
            errors.append(f"No SERVICE_SPECS entry for {boto_id}")
            continue
        c = boto3.client(boto_id, region_name="us-east-1")
        for op in ops:
            snake = boto_method_name(boto_id, op)
            if not hasattr(c, snake):
                errors.append(
                    f"{boto_id}.{op} -> {snake} missing on boto client"
                )
    return errors


def main() -> None:
    errs = validate_against_boto()
    if errs:
        print("Validation errors:")
        for e in errs:
            print(" ", e)
        raise SystemExit(1)

    for boto_id, ops in OPS.items():
        if boto_id not in SERVICE_SPECS:
            continue
        module_stem, class_name = SERVICE_SPECS[boto_id]
        path = PROVIDER / f"{module_stem}.py"
        path.write_text(
            render_file(boto_id, module_stem, class_name, ops), encoding="utf-8"
        )
        print("Wrote", path.relative_to(ROOT))

    print("Done.")


if __name__ == "__main__":
    main()

	# Generate explicit BaseService subclasses under :
	# server/providers/aws/*_service.py from AWS_SERVICE_OPERATIONS


	from __future__ import annotations

	import sys
	from pathlib import Path
	from textwrap import dedent

	from cloudenv.server.providers.aws.base import pascal_to_snake
	from cloudenv.server.providers.aws.operations_data import (
	AWS_SERVICE_OPERATIONS as OPS,
	)

	ROOT = Path(__file__).resolve().parents[1]
	PROVIDER = ROOT / "server" / "providers" / "aws"

	sys.path.insert(0, str(ROOT))
	sys.path.insert(0, str(Path(__file__).resolve().parent))

	# boto client id -> (module stem without _service, Python class name)
	SERVICE_SPECS: dict[str, tuple[str, str]] = {
	"secretsmanager": ("secrets_manager_service", "SecretsManagerService"),
	"ssm": ("ssm_service", "SSMService"),
	"logs": ("cloudwatch_logs_service", "CloudWatchLogsService"),
	"events": ("eventbridge_service", "EventBridgeService"),
	"kinesis": ("kinesis_service", "KinesisService"),
	"cloudwatch": ("cloudwatch_service", "CloudWatchService"),
	"stepfunctions": ("stepfunctions_service", "StepFunctionsService"),
	"firehose": ("firehose_service", "FirehoseService"),
	"ecs": ("ecs_service", "ECSService"),
	"glue": ("glue_service", "GlueService"),
	"athena": ("athena_service", "AthenaService"),
	"ses": ("ses_service", "SESService"),
	"sesv2": ("sesv2_service", "SESv2Service"),
	"acm": ("acm_service", "ACMService"),
	"wafv2": ("wafv2_service", "WAFv2Service"),
	"route53": ("route53_service", "Route53Service"),
	"rds": ("rds_service", "RDSService"),
	"elasticache": ("elasticache_service", "ElastiCacheService"),
	"sqs": ("sqs_service", "SQSService"),
	"dynamodb": ("dynamodb_service", "DynamoDBService"),
	"lambda": ("lambda_service", "LambdaService"),
	"iam": ("iam_service", "IAMService"),
	"s3": ("s3_service", "S3Service"),
	"ec2": ("ec2_service", "EC2Service"),
	"apigateway": ("apigateway_service", "APIGatewayService"),
	"apigatewayv2": ("apigatewayv2_service", "APIGatewayV2Service"),
	"cognito-idp": ("cognito_idp_service", "CognitoIDPService"),
	"cognito-identity": ("cognito_identity_service", "CognitoIdentityService"),
	}

	# (boto_id, OperationName) -> boto method name when it differs from pascal_to_snake(OperationName)
	METHOD_OVERRIDES: dict[tuple[str, str], str] = {
	# pascal_to_snake turns trailing ACLs into ac_ls; boto uses acls
	("wafv2", "ListWebACLs"): "list_web_acls",
	}


	def boto_method_name(boto_id: str, operation: str) -> str:
	return METHOD_OVERRIDES.get(
	(boto_id, operation), pascal_to_snake(operation)
	)


	def render_methods(boto_id: str, operations: list[str]) -> str:
	lines: list[str] = []
	for op in operations:
	snake = boto_method_name(boto_id, op)
	lines.append(
	f" def {op}(self, **kwargs):\n"
	f' """Forward to boto3 {snake} (IAM: {boto_id}:{op})."""\n'
	f" return self.client.{snake}(**kwargs)\n"
	)
	return "\n".join(lines)


	def render_lifecycle(boto_id: str, class_name: str) -> str:
	"""reset / health / describe_state blocks."""
	if boto_id == "s3":
	return dedent(
	"""
	def reset(self) -> None:
	try:
	buckets = self.client.list_buckets().get("Buckets", [])
	for b in buckets:
	name = b["Name"]
	objs = self.client.list_objects_v2(Bucket=name).get("Contents", [])
	for obj in objs:
	self.client.delete_object(Bucket=name, Key=obj["Key"])
	self.client.delete_bucket(Bucket=name)
	except Exception:
	pass

	def health(self) -> float:
	try:
	self.client.list_buckets()
	return 1.0
	except Exception:
	return 0.0

	def describe_state(self) -> dict:
	try:
	buckets = self.client.list_buckets().get("Buckets", [])
	result = []
	for b in buckets:
	name = b["Name"]
	objs = self.client.list_objects_v2(Bucket=name).get("Contents", [])
	result.append({"name": name, "object_count": len(objs)})
	return {"buckets": result}
	except Exception:
	return {"buckets": []}
	"""
	)

	if boto_id == "sqs":
	return dedent(
	"""
	def reset(self) -> None:
	try:
	queues = self.client.list_queues().get("QueueUrls", [])
	for q in queues:
	self.client.delete_queue(QueueUrl=q)
	except Exception:
	pass

	def health(self) -> float:
	try:
	self.client.list_queues()
	return 1.0
	except Exception:
	return 0.0

	def describe_state(self) -> dict:
	try:
	urls = self.client.list_queues().get("QueueUrls", [])
	queues = []
	for q in urls:
	attrs = self.client.get_queue_attributes(
	QueueUrl=q, AttributeNames=["ApproximateNumberOfMessages"]
	)["Attributes"]
	queues.append(
	{"url": q, "messages": int(attrs.get("ApproximateNumberOfMessages", 0))}
	)
	return {"queues": queues}
	except Exception:
	return {"queues": []}
	"""
	)

	if boto_id == "dynamodb":
	return dedent(
	"""
	def reset(self) -> None:
	try:
	tables = self.client.list_tables().get("TableNames", [])
	for t in tables:
	self.client.delete_table(TableName=t)
	except Exception:
	pass

	def health(self) -> float:
	try:
	self.client.list_tables()
	return 1.0
	except Exception:
	return 0.0

	def describe_state(self) -> dict:
	try:
	tables = self.client.list_tables().get("TableNames", [])
	return {"tables": tables}
	except Exception:
	return {"tables": []}
	"""
	)

	if boto_id == "lambda":
	return dedent(
	"""
	def reset(self) -> None:
	try:
	funcs = self.client.list_functions().get("Functions", [])
	for f in funcs:
	self.client.delete_function(FunctionName=f["FunctionName"])
	except Exception:
	pass

	def health(self) -> float:
	try:
	self.client.list_functions()
	return 1.0
	except Exception:
	return 0.0

	def describe_state(self) -> dict:
	try:
	funcs = self.client.list_functions().get("Functions", [])
	return {"functions": [f["FunctionName"] for f in funcs]}
	except Exception:
	return {"functions": []}
	"""
	)

	if boto_id == "iam":
	return dedent(
	"""
	def reset(self) -> None:
	try:
	policies = self.client.list_policies(Scope="Local").get("Policies", [])
	for p in policies:
	try:
	self.client.delete_policy(PolicyArn=p["Arn"])
	except Exception:
	pass
	roles = self.client.list_roles().get("Roles", [])
	for r in roles:
	try:
	self.client.delete_role(RoleName=r["RoleName"])
	except Exception:
	pass
	except Exception:
	pass

	def health(self) -> float:
	try:
	self.client.list_roles()
	return 1.0
	except Exception:
	return 0.0

	def describe_state(self) -> dict:
	try:
	roles = self.client.list_roles().get("Roles", [])
	return {"roles": [r["RoleName"] for r in roles]}
	except Exception:
	return {"roles": []}
	"""
	)

	if boto_id == "secretsmanager":
	return dedent(
	"""
	def reset(self) -> None:
	try:
	for s in self.client.list_secrets().get("SecretList", []):
	try:
	self.client.delete_secret(
	SecretId=s["ARN"], ForceDeleteWithoutRecovery=True
	)
	except Exception:
	pass
	except Exception:
	pass

	def health(self) -> float:
	try:
	self.client.list_secrets(MaxResults=1)
	return 1.0
	except Exception:
	return 0.0

	def describe_state(self) -> dict:
	try:
	n = len(self.client.list_secrets().get("SecretList", []))
	return {"secrets": n}
	except Exception:
	return {"secrets": 0}
	"""
	)

	if boto_id == "ssm":
	return dedent(
	"""
	def reset(self) -> None:
	try:
	paginator = self.client.get_paginator("describe_parameters")
	names = []
	for page in paginator.paginate():
	for p in page.get("Parameters", []):
	names.append(p["Name"])
	for name in names:
	try:
	self.client.delete_parameter(Name=name)
	except Exception:
	pass
	except Exception:
	pass

	def health(self) -> float:
	try:
	self.client.describe_parameters(MaxResults=1)
	return 1.0
	except Exception:
	return 0.0

	def describe_state(self) -> dict:
	try:
	r = self.client.describe_parameters(MaxResults=50)
	return {"parameters": len(r.get("Parameters", []))}
	except Exception:
	return {"parameters": 0}
	"""
	)

	if boto_id == "logs":
	return dedent(
	"""
	def reset(self) -> None:
	try:
	groups = self.client.describe_log_groups().get("logGroups", [])
	for g in groups:
	try:
	self.client.delete_log_group(logGroupName=g["logGroupName"])
	except Exception:
	pass
	except Exception:
	pass

	def health(self) -> float:
	try:
	self.client.describe_log_groups(limit=1)
	return 1.0
	except Exception:
	return 0.0

	def describe_state(self) -> dict:
	try:
	groups = self.client.describe_log_groups().get("logGroups", [])
	return {"log_groups": len(groups)}
	except Exception:
	return {"log_groups": 0}
	"""
	)

	if boto_id == "events":
	return dedent(
	"""
	def reset(self) -> None:
	try:
	buses = self.client.list_event_buses().get("EventBuses", [])
	for b in buses:
	name = b["Name"]
	if name == "default":
	continue
	try:
	self.client.delete_event_bus(Name=name)
	except Exception:
	pass
	except Exception:
	pass

	def health(self) -> float:
	try:
	self.client.list_event_buses(Limit=1)
	return 1.0
	except Exception:
	return 0.0

	def describe_state(self) -> dict:
	try:
	buses = self.client.list_event_buses().get("EventBuses", [])
	return {"event_buses": len(buses)}
	except Exception:
	return {"event_buses": 0}
	"""
	)

	if boto_id == "kinesis":
	return dedent(
	"""
	def reset(self) -> None:
	try:
	names = self.client.list_streams().get("StreamNames", [])
	for n in names:
	try:
	self.client.delete_stream(StreamName=n, EnforceConsumerDeletion=True)
	except Exception:
	pass
	except Exception:
	pass

	def health(self) -> float:
	try:
	self.client.list_streams(Limit=1)
	return 1.0
	except Exception:
	return 0.0

	def describe_state(self) -> dict:
	try:
	names = self.client.list_streams().get("StreamNames", [])
	return {"streams": len(names)}
	except Exception:
	return {"streams": 0}
	"""
	)

	if boto_id == "cloudwatch":
	return dedent(
	"""
	def reset(self) -> None:
	try:
	alarms = self.client.describe_alarms().get("MetricAlarms", [])
	names = [a["AlarmName"] for a in alarms]
	if names:
	self.client.delete_alarms(AlarmNames=names)
	boards = self.client.list_dashboards().get("DashboardEntries", [])
	for d in boards:
	try:
	self.client.delete_dashboards(DashboardNames=[d["DashboardName"]])
	except Exception:
	pass
	except Exception:
	pass

	def health(self) -> float:
	try:
	self.client.list_metrics(Namespace="AWS/EC2")
	return 1.0
	except Exception:
	return 0.0

	def describe_state(self) -> dict:
	try:
	alarms = self.client.describe_alarms().get("MetricAlarms", [])
	return {"alarms": len(alarms)}
	except Exception:
	return {"alarms": 0}
	"""
	)

	if boto_id == "stepfunctions":
	return dedent(
	"""
	def reset(self) -> None:
	try:
	sms = self.client.list_state_machines().get("stateMachines", [])
	for sm in sms:
	try:
	self.client.delete_state_machine(stateMachineArn=sm["stateMachineArn"])
	except Exception:
	pass
	except Exception:
	pass

	def health(self) -> float:
	try:
	self.client.list_state_machines(maxResults=1)
	return 1.0
	except Exception:
	return 0.0

	def describe_state(self) -> dict:
	try:
	sms = self.client.list_state_machines().get("stateMachines", [])
	return {"state_machines": len(sms)}
	except Exception:
	return {"state_machines": 0}
	"""
	)

	if boto_id == "firehose":
	return dedent(
	"""
	def reset(self) -> None:
	try:
	names = self.client.list_delivery_streams().get("DeliveryStreamNames", [])
	for n in names:
	try:
	self.client.delete_delivery_stream(DeliveryStreamName=n)
	except Exception:
	pass
	except Exception:
	pass

	def health(self) -> float:
	try:
	self.client.list_delivery_streams(Limit=1)
	return 1.0
	except Exception:
	return 0.0

	def describe_state(self) -> dict:
	try:
	names = self.client.list_delivery_streams().get("DeliveryStreamNames", [])
	return {"delivery_streams": len(names)}
	except Exception:
	return {"delivery_streams": 0}
	"""
	)

	if boto_id == "ecs":
	return dedent(
	"""
	def reset(self) -> None:
	try:
	clusters = self.client.list_clusters().get("clusterArns", [])
	for arn in clusters:
	try:
	self.client.delete_cluster(cluster=arn)
	except Exception:
	pass
	except Exception:
	pass

	def health(self) -> float:
	try:
	self.client.list_clusters(maxResults=1)
	return 1.0
	except Exception:
	return 0.0

	def describe_state(self) -> dict:
	try:
	clusters = self.client.list_clusters().get("clusterArns", [])
	return {"clusters": len(clusters)}
	except Exception:
	return {"clusters": 0}
	"""
	)

	if boto_id == "glue":
	return dedent(
	"""
	def reset(self) -> None:
	try:
	dbs = self.client.get_databases().get("DatabaseList", [])
	for db in dbs:
	name = db["Name"]
	if name == "default":
	continue
	try:
	self.client.delete_database(Name=name)
	except Exception:
	pass
	except Exception:
	pass

	def health(self) -> float:
	try:
	self.client.get_databases(MaxResults=1)
	return 1.0
	except Exception:
	return 0.0

	def describe_state(self) -> dict:
	try:
	dbs = self.client.get_databases().get("DatabaseList", [])
	return {"databases": len(dbs)}
	except Exception:
	return {"databases": 0}
	"""
	)

	if boto_id == "athena":
	return dedent(
	"""
	def reset(self) -> None:
	try:
	wgs = self.client.list_work_groups().get("WorkGroups", [])
	for wg in wgs:
	name = wg["Name"]
	if name == "primary":
	continue
	try:
	self.client.delete_work_group(WorkGroup=name, RecursiveDeleteOption=True)
	except Exception:
	pass
	except Exception:
	pass

	def health(self) -> float:
	try:
	self.client.list_work_groups(MaxResults=1)
	return 1.0
	except Exception:
	return 0.0

	def describe_state(self) -> dict:
	try:
	wgs = self.client.list_work_groups().get("WorkGroups", [])
	return {"work_groups": len(wgs)}
	except Exception:
	return {"work_groups": 0}
	"""
	)

	if boto_id in ("ses", "sesv2"):
	return dedent(
	"""
	def reset(self) -> None:
	pass

	def health(self) -> float:
	try:
	if hasattr(self.client, "list_identities"):
	self.client.list_identities(MaxItems=1)
	else:
	self.client.list_email_identities(MaxResults=1)
	return 1.0
	except Exception:
	return 0.0

	def describe_state(self) -> dict:
	return {}
	"""
	)

	if boto_id == "acm":
	return dedent(
	"""
	def reset(self) -> None:
	try:
	certs = self.client.list_certificates().get("CertificateSummaryList", [])
	for c in certs:
	try:
	self.client.delete_certificate(CertificateArn=c["CertificateArn"])
	except Exception:
	pass
	except Exception:
	pass

	def health(self) -> float:
	try:
	self.client.list_certificates(MaxItems=1)
	return 1.0
	except Exception:
	return 0.0

	def describe_state(self) -> dict:
	try:
	certs = self.client.list_certificates().get("CertificateSummaryList", [])
	return {"certificates": len(certs)}
	except Exception:
	return {"certificates": 0}
	"""
	)

	if boto_id == "wafv2":
	return dedent(
	"""
	def reset(self) -> None:
	pass

	def health(self) -> float:
	try:
	self.client.list_web_acls(Scope="REGIONAL", Limit=1)
	return 1.0
	except Exception:
	return 0.0

	def describe_state(self) -> dict:
	return {}
	"""
	)

	if boto_id == "route53":
	return dedent(
	"""
	def reset(self) -> None:
	try:
	zones = self.client.list_hosted_zones().get("HostedZones", [])
	for z in zones:
	try:
	self.client.delete_hosted_zone(Id=z["Id"])
	except Exception:
	pass
	except Exception:
	pass

	def health(self) -> float:
	try:
	self.client.list_hosted_zones(MaxItems="1")
	return 1.0
	except Exception:
	return 0.0

	def describe_state(self) -> dict:
	try:
	zones = self.client.list_hosted_zones().get("HostedZones", [])
	return {"hosted_zones": len(zones)}
	except Exception:
	return {"hosted_zones": 0}
	"""
	)

	if boto_id == "rds":
	return dedent(
	"""
	def reset(self) -> None:
	try:
	dbs = self.client.describe_db_instances().get("DBInstances", [])
	for db in dbs:
	ident = db["DBInstanceIdentifier"]
	try:
	self.client.delete_db_instance(
	DBInstanceIdentifier=ident, SkipFinalSnapshot=True
	)
	except Exception:
	pass
	except Exception:
	pass

	def health(self) -> float:
	try:
	self.client.describe_db_instances(MaxRecords=20)
	return 1.0
	except Exception:
	return 0.0

	def describe_state(self) -> dict:
	try:
	dbs = self.client.describe_db_instances().get("DBInstances", [])
	return {"db_instances": len(dbs)}
	except Exception:
	return {"db_instances": 0}
	"""
	)

	if boto_id == "elasticache":
	return dedent(
	"""
	def reset(self) -> None:
	try:
	clusters = self.client.describe_cache_clusters().get("CacheClusters", [])
	for c in clusters:
	cid = c["CacheClusterId"]
	try:
	self.client.delete_cache_cluster(CacheClusterId=cid)
	except Exception:
	pass
	except Exception:
	pass

	def health(self) -> float:
	try:
	self.client.describe_cache_clusters(MaxRecords=20)
	return 1.0
	except Exception:
	return 0.0

	def describe_state(self) -> dict:
	try:
	clusters = self.client.describe_cache_clusters().get("CacheClusters", [])
	return {"cache_clusters": len(clusters)}
	except Exception:
	return {"cache_clusters": 0}
	"""
	)

	if boto_id == "ec2":
	return dedent(
	"""
	def reset(self) -> None:
	try:
	res = self.client.describe_instances()
	for r in res.get("Reservations", []):
	for inst in r.get("Instances", []):
	iid = inst["InstanceId"]
	try:
	self.client.terminate_instances(InstanceIds=[iid])
	except Exception:
	pass
	except Exception:
	pass

	def health(self) -> float:
	try:
	self.client.describe_regions()
	return 1.0
	except Exception:
	return 0.0

	def describe_state(self) -> dict:
	try:
	res = self.client.describe_instances()
	n = sum(len(r.get("Instances", [])) for r in res.get("Reservations", []))
	return {"instances": n}
	except Exception:
	return {"instances": 0}
	"""
	)

	if boto_id == "apigateway":
	return dedent(
	"""
	def reset(self) -> None:
	try:
	apis = self.client.get_rest_apis().get("items", [])
	for a in apis:
	try:
	self.client.delete_rest_api(restApiId=a["id"])
	except Exception:
	pass
	except Exception:
	pass

	def health(self) -> float:
	try:
	self.client.get_rest_apis(limit=1)
	return 1.0
	except Exception:
	return 0.0

	def describe_state(self) -> dict:
	try:
	apis = self.client.get_rest_apis().get("items", [])
	return {"rest_apis": len(apis)}
	except Exception:
	return {"rest_apis": 0}
	"""
	)

	if boto_id == "apigatewayv2":
	return dedent(
	"""
	def reset(self) -> None:
	try:
	apis = self.client.get_apis().get("Items", [])
	for a in apis:
	try:
	self.client.delete_api(ApiId=a["ApiId"])
	except Exception:
	pass
	except Exception:
	pass

	def health(self) -> float:
	try:
	self.client.get_apis(MaxResults="1")
	return 1.0
	except Exception:
	return 0.0

	def describe_state(self) -> dict:
	try:
	apis = self.client.get_apis().get("Items", [])
	return {"apis": len(apis)}
	except Exception:
	return {"apis": 0}
	"""
	)

	if boto_id == "cognito-idp":
	return dedent(
	"""
	def reset(self) -> None:
	try:
	pools = self.client.list_user_pools(MaxResults=60).get("UserPools", [])
	for p in pools:
	try:
	self.client.delete_user_pool(UserPoolId=p["Id"])
	except Exception:
	pass
	except Exception:
	pass

	def health(self) -> float:
	try:
	self.client.list_user_pools(MaxResults=1)
	return 1.0
	except Exception:
	return 0.0

	def describe_state(self) -> dict:
	try:
	pools = self.client.list_user_pools(MaxResults=60).get("UserPools", [])
	return {"user_pools": len(pools)}
	except Exception:
	return {"user_pools": 0}
	"""
	)

	if boto_id == "cognito-identity":
	return dedent(
	"""
	def reset(self) -> None:
	try:
	pools = self.client.list_identity_pools(MaxResults=60).get("IdentityPools", [])
	for p in pools:
	try:
	self.client.delete_identity_pool(IdentityPoolId=p["IdentityPoolId"])
	except Exception:
	pass
	except Exception:
	pass

	def health(self) -> float:
	try:
	self.client.list_identity_pools(MaxResults=1)
	return 1.0
	except Exception:
	return 0.0

	def describe_state(self) -> dict:
	try:
	pools = self.client.list_identity_pools(MaxResults=60).get("IdentityPools", [])
	return {"identity_pools": len(pools)}
	except Exception:
	return {"identity_pools": 0}
	"""
	)

	# default
	return dedent(
	"""
	def reset(self) -> None:
	pass

	def health(self) -> float:
	return 1.0

	def describe_state(self) -> dict:
	return {}
	"""
	)


	def render_file(
	boto_id: str, module_stem: str, class_name: str, operations: list[str]
	) -> str:
	methods = render_methods(boto_id, operations)
	life = render_lifecycle(boto_id, class_name)
	doc = f'"""Explicit {class_name} — boto `{boto_id}` (IAM: `{boto_id}:<Operation>`)."""'
	return dedent(
	f'''\
	{doc}
	from __future__ import annotations

	from botocore.client import BaseClient

	from cloudenv.server.providers.aws.client import get_client
	from cloudenv.server.providers.aws.base import BaseService


	class {class_name}(BaseService):
	def __init__(self) -> None:
	self.client: BaseClient = get_client("{boto_id}")

	{methods}
	{life}
	'''
	)


	def validate_against_boto() -> list[str]:
	import boto3

	errors: list[str] = []
	for boto_id, ops in OPS.items():
	if boto_id not in SERVICE_SPECS:
	errors.append(f"No SERVICE_SPECS entry for {boto_id}")
	continue
	c = boto3.client(boto_id, region_name="us-east-1")
	for op in ops:
	snake = boto_method_name(boto_id, op)
	if not hasattr(c, snake):
	errors.append(
	f"{boto_id}.{op} -> {snake} missing on boto client"
	)
	return errors


	def main() -> None:
	errs = validate_against_boto()
	if errs:
	print("Validation errors:")
	for e in errs:
	print(" ", e)
	raise SystemExit(1)

	for boto_id, ops in OPS.items():
	if boto_id not in SERVICE_SPECS:
	continue
	module_stem, class_name = SERVICE_SPECS[boto_id]
	path = PROVIDER / f"{module_stem}.py"
	path.write_text(
	render_file(boto_id, module_stem, class_name, ops), encoding="utf-8"
	)
	print("Wrote", path.relative_to(ROOT))

	print("Done.")


	if __name__ == "__main__":
	main()