AWS Interview Questions for Data Engineers, Part 2: The boto3 Coding Round

import os
os.environ.update(AWS_ACCESS_KEY_ID="testing", AWS_SECRET_ACCESS_KEY="testing", AWS_DEFAULT_REGION="ap-south-1")
import boto3
from botocore.exceptions import ClientError
from moto import mock_aws

with mock_aws():                                     # local AWS emulator: no real account, no cost
    s3 = boto3.client("s3", region_name="ap-south-1")
    s3.create_bucket(Bucket="lake", CreateBucketConfiguration={"LocationConstraint": "ap-south-1"})
    for i in range(2500):
        s3.put_object(Bucket="lake", Key=f"raw/orders/part-{i:05d}.json", Body=b"{}")

    # Task 1: "How many files are under raw/orders/?"
    first_call = s3.list_objects_v2(Bucket="lake", Prefix="raw/orders/")
    pages = s3.get_paginator("list_objects_v2").paginate(Bucket="lake", Prefix="raw/orders/")
    total = sum(page["KeyCount"] for page in pages)
    print(f"T1 one list_objects_v2 call: {first_call['KeyCount']} keys, IsTruncated={first_call['IsTruncated']} | paginator: {total} keys")

    # Task 2: "Make the daily job safe to retry"
    def run_job(day, run_id, deterministic):
        key = f"curated/orders/day={day}/part-0.json" if deterministic else f"curated/orders/day={day}/{run_id}.json"
        s3.put_object(Bucket="lake", Key=key, Body=b'{"rows": 120}')
    for deterministic in (False, True):
        for run_id in ("run-a", "run-b-retry"):              # the scheduler retried the same day
            run_job("2026-10-09" if deterministic else "2026-10-08", run_id, deterministic)
    for day, label in (("2026-10-08", "key per run   "), ("2026-10-09", "key per day   ")):
        n = s3.list_objects_v2(Bucket="lake", Prefix=f"curated/orders/day={day}/")["KeyCount"]
        print(f"T2 {label}: {n} output file(s) after one retry")

    # Task 3: "Record each payment event exactly once"
    ddb = boto3.client("dynamodb", region_name="ap-south-1")
    ddb.create_table(TableName="payments", KeySchema=[{"AttributeName": "event_id", "KeyType": "HASH"}],
                     AttributeDefinitions=[{"AttributeName": "event_id", "AttributeType": "S"}], BillingMode="PAY_PER_REQUEST")
    outcomes = []
    for attempt in range(2):                                  # the same event delivered twice
        try:
            ddb.put_item(TableName="payments", Item={"event_id": {"S": "evt-981"}, "amount": {"N": "499"}},
                         ConditionExpression="attribute_not_exists(event_id)")
            outcomes.append("written")
        except ClientError as err:
            outcomes.append(err.response["Error"]["Code"])
    print(f"T3 conditional put, delivered twice: {outcomes} | items stored: {ddb.scan(TableName='payments')['Count']}")

    # Task 4: "Delete everything under raw/orders/"
    requests = 0
    for page in s3.get_paginator("list_objects_v2").paginate(Bucket="lake", Prefix="raw/orders/"):
        batch = [{"Key": o["Key"]} for o in page.get("Contents", [])]   # each page holds at most 1,000 keys
        if batch:
            s3.delete_objects(Bucket="lake", Delete={"Objects": batch}); requests += 1
    left = s3.list_objects_v2(Bucket="lake", Prefix="raw/orders/")["KeyCount"]
    print(f"T4 delete page by page: {requests} delete_objects requests, {left} keys left")
    for i in range(1500):
        s3.put_object(Bucket="lake", Key=f"tmp/{i:05d}", Body=b"")
    too_many = [{"Key": f"tmp/{i:05d}"} for i in range(1500)]
    try:
        s3.delete_objects(Bucket="lake", Delete={"Objects": too_many})
        print("   the emulator also accepted one request with 1,500 keys")
    except ClientError as err:
        print(f"   one request with 1,500 keys: {err.response['Error']['Code']}")

    # Task 5: "Find the latest daily file without reading them all"
    for day in (8, 9, 10, 11):
        s3.put_object(Bucket="lake", Key=f"exports/day={day}/data.csv", Body=b"x")
        s3.put_object(Bucket="lake", Key=f"exports2/day=2026-10-{day:02d}/data.csv", Body=b"x")
    for prefix in ("exports/", "exports2/"):
        listed = [o["Key"] for o in s3.list_objects_v2(Bucket="lake", Prefix=prefix)["Contents"]]
        print(f"T5 keys under {prefix:<9} listed in order: {[k.split('/')[1] for k in listed]} -> last is {listed[-1].split('/')[1]}")

T1 one list_objects_v2 call: 1000 keys, IsTruncated=True | paginator: 2500 keys
T2 key per run   : 2 output file(s) after one retry
T2 key per day   : 1 output file(s) after one retry
T3 conditional put, delivered twice: ['written', 'ConditionalCheckFailedException'] | items stored: 1
T4 delete page by page: 3 delete_objects requests, 0 keys left
   the emulator also accepted one request with 1,500 keys
T5 keys under exports/  listed in order: ['day=10', 'day=11', 'day=8', 'day=9'] -> last is day=9
T5 keys under exports2/ listed in order: ['day=2026-10-08', 'day=2026-10-09', 'day=2026-10-10', 'day=2026-10-11'] -> last is day=2026-10-11