Solution code

Solution code is your reference implementation for a step. It demonstrates how to complete the step correctly by creating the AWS resources the student is expected to create. The platform runs it during testing to validate the project works end-to-end. Students never see or run this code — they write their own.

One solution file per step, located at solution/boto3/{objective_id}_{step_id}.py.

Function signature

Every solution file must define exactly this function:

async def run(session, context, logger):

Parameters

session — a boto3 Session scoped to the sandbox account. Use it to create service clients:

s3 = session.client("s3")
ec2 = session.client("ec2")

context — a dict containing:

• resolved — list of validated handles from prior steps: [{"name": "01_01_bucket", "type": "AWS::S3::Bucket", "id": "my-bucket-a3f9c2d1"}, ...]
• spec — the step’s expected inputs and outputs from the project spec

For the first step in a project, resolved will be empty. For subsequent steps, it contains the validated outputs from all prior steps.

logger — standard Python logger. Use it to record what the function creates or encounters.

Return format

On success, return a dict with success: True and a resources list. Each resource entry maps to one of the step’s declared outputs:

return {
    "success": True,
    "resources": [
        {"name": "01_01_bucket", "type": "AWS::S3::Bucket", "id": "01-01-bucket-a3f9c2d1"}
    ]
}

• name — the handle name from the step’s outputs declaration in spec.yaml
• type — the CloudFormation resource type, or "string" for non-resource outputs
• id — the actual AWS resource identifier or value

On failure, return an error dict:

return {"error": str(e), "code": e.response["Error"]["Code"]}

Never raise exceptions — always catch and return a dict.

Import constraints

Only these imports are allowed:

• boto3 and boto3 sub-modules
• botocore.exceptions.ClientError
• Standard library modules (os, json, uuid, datetime, etc.)

Don’t import third-party packages or internal SDK modules.

Resource naming

Use {obj_id}-{step_id}-{resource}-{uuid.uuid4().hex[:8]} for resource names to avoid collisions across test runs:

bucket_name = f"01-01-bucket-{uuid.uuid4().hex[:8]}"

Wait for resource availability

Many AWS resources aren’t immediately available after creation. Always wait for resources to reach their ready state before returning — subsequent steps depend on them being fully operational.

Use boto3 waiters where available:

# VPC
ec2.get_waiter("vpc_available").wait(VpcIds=[vpc_id])

# NAT Gateway (can take 2–5 minutes)
ec2.get_waiter("nat_gateway_available").wait(
    NatGatewayIds=[nat_gw_id],
    WaiterConfig={"Delay": 15, "MaxAttempts": 40},
)

# Load Balancer
elbv2.get_waiter("load_balancer_available").wait(LoadBalancerArns=[alb_arn])

Skipping waiters causes cascading failures in downstream steps.

Worked example — first step

Spec declares: outputs: [{name: "01_01_bucket", type: "AWS::S3::Bucket"}]

import uuid
from botocore.exceptions import ClientError


async def run(session, context, logger):
    s3 = session.client("s3")
    bucket_name = f"01-01-bucket-{uuid.uuid4().hex[:8]}"
    try:
        s3.create_bucket(Bucket=bucket_name)
        logger.info(f"Created bucket: {bucket_name}")
        return {
            "success": True,
            "resources": [
                {"name": "01_01_bucket", "type": "AWS::S3::Bucket", "id": bucket_name}
            ],
        }
    except ClientError as e:
        logger.error(f"Failed to create bucket: {e}")
        return {"error": str(e), "code": e.response["Error"]["Code"]}

Worked example — subsequent step consuming a prior handle

Spec declares: inputs: ["01_01_bucket"], outputs: [{name: "01_02_policy", type: "AWS::S3::BucketPolicy"}]

import json
from botocore.exceptions import ClientError


async def run(session, context, logger):
    s3 = session.client("s3")
    resolved = {r["name"]: r for r in context.get("resolved", [])}

    bucket_handle = resolved.get("01_01_bucket")
    if not bucket_handle:
        return {"error": "Prior dependency 01_01_bucket not resolved", "code": "MISSING_INPUT"}

    bucket_name = bucket_handle["id"]
    policy = {
        "Version": "2012-10-17",
        "Statement": [{
            "Sid": "PublicReadGetObject",
            "Effect": "Allow",
            "Principal": "*",
            "Action": "s3:GetObject",
            "Resource": f"arn:aws:s3:::{bucket_name}/*",
        }],
    }
    try:
        s3.put_bucket_policy(Bucket=bucket_name, Policy=json.dumps(policy))
        logger.info(f"Applied bucket policy to {bucket_name}")
        return {
            "success": True,
            "resources": [
                {"name": "01_02_policy", "type": "AWS::S3::BucketPolicy", "id": bucket_name}
            ],
        }
    except ClientError as e:
        logger.error(f"Failed to apply bucket policy: {e}")
        return {"error": str(e), "code": e.response["Error"]["Code"]}

Look up prior handles by name from context["resolved"], check they’re present before using them, and return an error dict if a dependency is missing.