Amazon EC2 — boto3 in Python¶

CFG = Config(retries={"mode": "standard", "max_attempts": 5}, connect_timeout=5, read_timeout=30)
ec2 = boto3.client("ec2", config=CFG)                # one client per Region, reuse it; EC2 clients are thread-safe

# 1. Let EC2 resolve the alias at launch (needs ssm:GetParameters on the parameter)
ALIAS = "resolve:ssm:/aws/service/ami-amazon-linux-latest/al2023-ami-kernel-default-arm64"
# 2. Or resolve it yourself, to log the exact AMI ID the test used
ssm = boto3.client("ssm", config=CFG)
ami_id = ssm.get_parameter(Name="/aws/service/ami-amazon-linux-latest/al2023-ami-kernel-default-arm64")["Parameter"]["Value"]
# 3. Your own golden images: newest first by CreationDate
mine = ec2.describe_images(Owners=["self"], Filters=[{"Name": "tag:app", "Values": ["web"]}, {"Name": "state", "Values": ["available"]}])["Images"]
newest = max(mine, key=lambda i: i["CreationDate"], default=None)

"""Build and run EC2 launch requests. boto3 calls are kept thin; the request shape is a pure function."""
import hashlib
from collections.abc import Mapping, Sequence

from botocore.exceptions import ClientError

AL2023_PARAM = "/aws/service/ami-amazon-linux-latest/al2023-ami-kernel-default-{arch}"   # arch: x86_64 | arm64


def latest_al2023_ami(ssm, arch: str = "x86_64") -> str:
    """Resolve the current AMI ID from the public SSM parameter (never hard-code AMI IDs)."""
    return ssm.get_parameter(Name=AL2023_PARAM.format(arch=arch))["Parameter"]["Value"]


def tag_list(tags: Mapping[str, str]) -> list[dict]:
    return [{"Key": k, "Value": v} for k, v in sorted(tags.items())]


def client_token(run_id: str, name: str) -> str:
    """Deterministic token: a retried launch of the same logical instance cannot create a second one."""
    return hashlib.sha256(f"{run_id}:{name}".encode()).hexdigest()[:32]       # <= 64 ASCII characters


def build_run_params(
    *, ami_id: str, instance_type: str, subnet_id: str, security_group_ids: Sequence[str],
    tags: Mapping[str, str], run_id: str, name: str, user_data: str | None = None,
    profile_name: str | None = None, root_gb: int = 20,
) -> dict:
    """The exact RunInstances request. Pure, so the policy (IMDSv2, tags, encryption) is unit-testable."""
    params: dict = {
        "ImageId": ami_id,
        "InstanceType": instance_type,
        "MinCount": 1,
        "MaxCount": 1,
        "SubnetId": subnet_id,
        "SecurityGroupIds": list(security_group_ids),
        "MetadataOptions": {"HttpTokens": "required", "HttpEndpoint": "enabled", "HttpPutResponseHopLimit": 1},
        "BlockDeviceMappings": [{
            "DeviceName": "/dev/xvda",
            "Ebs": {"VolumeSize": root_gb, "VolumeType": "gp3", "Encrypted": True, "DeleteOnTermination": True},
        }],
        "TagSpecifications": [                      # tag at creation: no window where the resource is untagged
            {"ResourceType": rt, "Tags": tag_list({**tags, "Name": name, "run-id": run_id})}
            for rt in ("instance", "volume")
        ],
        "InstanceInitiatedShutdownBehavior": "terminate",   # `shutdown -h` inside the guest ends billing
        "ClientToken": client_token(run_id, name),
    }
    if user_data is not None:
        params["UserData"] = user_data              # plain text: botocore base64-encodes it for RunInstances
    if profile_name:
        params["IamInstanceProfile"] = {"Name": profile_name}
    return params


def is_error(exc: ClientError, *codes: str) -> bool:
    return exc.response.get("Error", {}).get("Code") in codes


def can_launch(ec2, **params) -> bool:
    """DryRun: True if IAM and parameters would allow the launch. Nothing is created and nothing is billed."""
    try:
        ec2.run_instances(DryRun=True, **params)
    except ClientError as exc:
        if is_error(exc, "DryRunOperation"):         # the "success" outcome of a dry run
            return True
        if is_error(exc, "UnauthorizedOperation"):
            return False
        raise
    raise AssertionError("DryRun call returned without an error")      # cannot happen on real EC2


def launch_with_fallback(ec2, params: dict, instance_types: Sequence[str]) -> dict:
    """Try instance types in order when the Region/AZ has no capacity for the first choice."""
    last: ClientError | None = None
    for itype in instance_types:
        try:
            return ec2.run_instances(**{**params, "InstanceType": itype})["Instances"][0]
        except ClientError as exc:
            if not is_error(exc, "InsufficientInstanceCapacity"):
                raise
            last = exc
    assert last is not None
    raise last

user_data = "#!/bin/bash\necho hello > /tmp/hello\n"
ec2.create_launch_template(
    LaunchTemplateName="web",
    LaunchTemplateData={
        "ImageId": ALIAS,
        "InstanceType": "t4g.nano",
        "UserData": base64.b64encode(user_data.encode()).decode(),     # launch templates do NOT base64-encode for you
        "MetadataOptions": {"HttpTokens": "required", "HttpPutResponseHopLimit": 2, "HttpEndpoint": "enabled"},
        "TagSpecifications": [{"ResourceType": "instance", "Tags": [{"Key": "app", "Value": "web"}]}],
    },
)
resp = ec2.run_instances(
    LaunchTemplate={"LaunchTemplateName": "web", "Version": "$Latest"},    # or LaunchTemplateId + a pinned Version
    MinCount=1, MaxCount=1,
    TagSpecifications=[{"ResourceType": "instance", "Tags": [{"Key": "run-id", "Value": "r-1"}]}],   # merged with the template's tags
)

pages = ec2.get_paginator("describe_instances").paginate(
    Filters=[{"Name": "tag:app", "Values": ["web"]}, {"Name": "instance-state-name", "Values": ["pending", "running"]}],
    PaginationConfig={"PageSize": 100},
)
ids = [i["InstanceId"] for page in pages for reservation in page["Reservations"] for i in reservation["Instances"]]

waiter_cfg = {"Delay": 5, "MaxAttempts": 60}         # 5 s x 60 = 5 minutes; the defaults are 15 s x 40
ec2.get_waiter("instance_running").wait(InstanceIds=ids, WaiterConfig=waiter_cfg)
print(ec2.stop_instances(InstanceIds=ids)["StoppingInstances"][0]["CurrentState"]["Name"])
ec2.get_waiter("instance_stopped").wait(InstanceIds=ids, WaiterConfig=waiter_cfg)
print(ec2.start_instances(InstanceIds=ids)["StartingInstances"][0]["CurrentState"]["Name"])
print(ec2.terminate_instances(InstanceIds=ids)["TerminatingInstances"][0]["CurrentState"]["Name"])

def launch_and_wait(ec2, params: dict, timeout_s: int = 300) -> str:
    instance_id = ec2.run_instances(**params)["Instances"][0]["InstanceId"]
    ec2.get_waiter("instance_running").wait(
        InstanceIds=[instance_id], WaiterConfig={"Delay": 5, "MaxAttempts": max(1, timeout_s // 5)}
    )
    return instance_id


def instances_by_run_id(ec2, run_id: str, states: Sequence[str] = ("pending", "running", "stopping", "stopped")) -> list[dict]:
    """Paginated, filtered describe: cheap on API quota and never returns terminated leftovers."""
    found = []
    pages = ec2.get_paginator("describe_instances").paginate(
        Filters=[{"Name": "tag:run-id", "Values": [run_id]}, {"Name": "instance-state-name", "Values": list(states)}]
    )
    for page in pages:
        for reservation in page["Reservations"]:
            found.extend(reservation["Instances"])
    return found


def terminate_run(ec2, run_id: str) -> list[str]:
    """Sweeper used in fixture teardown and by the scheduled janitor job."""
    ids = [i["InstanceId"] for i in instances_by_run_id(ec2, run_id)]
    if ids:
        ec2.terminate_instances(InstanceIds=ids)
    return ids

vpc_id = ec2.describe_vpcs(Filters=[{"Name": "isDefault", "Values": ["true"]}])["Vpcs"][0]["VpcId"]
web_sg = ec2.create_security_group(
    GroupName="app-web", Description="web tier", VpcId=vpc_id,
    TagSpecifications=[{"ResourceType": "security-group", "Tags": [{"Key": "Name", "Value": "app-web"}]}],
)["GroupId"]
ec2.authorize_security_group_ingress(GroupId=web_sg, IpPermissions=[
    {"IpProtocol": "tcp", "FromPort": 443, "ToPort": 443, "IpRanges": [{"CidrIp": "10.0.0.0/8", "Description": "corp network"}]},
])
db_sg = ec2.create_security_group(GroupName="app-db", Description="db tier", VpcId=vpc_id)["GroupId"]
ec2.authorize_security_group_ingress(GroupId=db_sg, IpPermissions=[
    {"IpProtocol": "tcp", "FromPort": 5432, "ToPort": 5432, "UserIdGroupPairs": [{"GroupId": web_sg}]},   # a group, not a CIDR
])

ec2.create_tags(Resources=[web_sg], Tags=[{"Key": "owner", "Value": "qa"}])     # overwrites the value if the key exists
ec2.delete_tags(Resources=[web_sg], Tags=[{"Key": "owner"}])                    # no Value: delete the key whatever its value

spot = ec2.run_instances(
    ImageId=ALIAS, InstanceType="t4g.nano", MinCount=1, MaxCount=1,
    InstanceMarketOptions={"MarketType": "spot",
                           "SpotOptions": {"SpotInstanceType": "one-time", "InstanceInterruptionBehavior": "terminate"}},
)["Instances"][0]
print("lifecycle:", spot["InstanceLifecycle"])                                  # "spot"; On-Demand instances have no such key

try:
    ec2.describe_instances(InstanceIds=["i-0123456789abcdef0"])
except ClientError as exc:                           # EC2 has NO modeled exception classes: always ClientError + Error.Code
    print("code:", exc.response["Error"]["Code"], "| http:", exc.response["ResponseMetadata"]["HTTPStatusCode"])
print("exception classes on the client:", [n for n in dir(ec2.exceptions) if not n.startswith("_")])

ec2r = boto3.resource("ec2", config=CFG)             # feature-frozen: fine for quick scripts, not for new features
inst = ec2r.create_instances(ImageId=ALIAS, InstanceType="t4g.nano", MinCount=1, MaxCount=1)[0]
inst.wait_until_running()
inst.reload()                                        # attributes are lazy: reload() refreshes them
print("resource:", inst.state["Name"], inst.instance_type)
inst.terminate()

own images found: 0 | SSM value looks like an AMI id: True
tags: ['app', 'run-id']
matching instances: 1
stopping
pending
shutting-down
db group allows the web group: True
lifecycle: spot
code: InvalidInstanceID.NotFound | http: 400
exception classes on the client: ['ClientError', 'from_code']
resource: running t4g.nano