Python Hands-On Labs

Prerequisites

Complete loops, functions, exceptions, and files in fundamentals. Use Python 3; no packages or cloud account are required. Work in a scratch directory for input and output files.

Sample input

Save as servers.csv:

name,server,status
web-01,prod,healthy
web-02,dev,warning
web-03,prod,critical

Task

Create filter_servers.py to read the file, keep critical records, and write critical-servers.csv. Validate headers, process one row at a time, and print the count. Invalid or missing input must produce an error and nonzero exit code. Do not replace an existing report without removing it first.

Run python filter_servers.py.

Expected output

The terminal prints Found 1 critical servers. and the file contains:

name,server,status
web-03,prod,critical

import csv
import sys
from pathlib import Path


def make_report(source, destination):
    if source.resolve() == destination.resolve():
        raise ValueError("Input and output must be different files")
    fields = ["name", "server", "status"]
    count = 0
    created = False
    try:
        with source.open(encoding="utf-8", newline="") as src:
            reader = csv.DictReader(src, strict=True)
            if reader.fieldnames != fields:
                raise ValueError("Expected headers: name,server,status")
            with destination.open("x", encoding="utf-8", newline="") as dst:
                created = True
                writer = csv.DictWriter(dst, fieldnames=fields)
                writer.writeheader()
                for row in reader:
                    if None in row or any(row[key] is None for key in fields):
                        raise ValueError("Unexpected number of fields")
                    if row["status"] == "critical":
                        writer.writerow(row)
                        count += 1
    except (OSError, ValueError, csv.Error, UnicodeError):
        if created:
            destination.unlink(missing_ok=True)
        raise
    return count


def main():
    try:
        count = make_report(Path("servers.csv"), Path("critical-servers.csv"))
    except (OSError, ValueError, csv.Error, UnicodeError) as error:
        print(f"Report failed: {error}", file=sys.stderr)
        return 1
    print(f"Found {count} critical servers.")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

Check your work

Use a fresh output path or remove the generated report between successful runs.

Inspect $LASTEXITCODE in PowerShell or echo $? in bash immediately after the command.

Knowledge check and challenge

Why stream records? Memory usage stays bounded by the current record instead of growing with the whole file.

Add input, output, and status flags using the production scripting bridge. Test a quoted comma in a name and a malformed row. Explain whether a scheduled job should replace or retain its previous report.

Prerequisites

Complete functions, exceptions, files, and modules in fundamentals, then the file-processing lab. This bridge runs locally without AWS credentials.

Set up a project environment

Create a project directory and run python -m venv .venv. Activate it with .venv\Scripts\Activate.ps1 in PowerShell or source .venv/bin/activate in bash. Alternatively, call .venv\Scripts\python.exe on Windows or .venv/bin/python on Unix directly. Confirm python -c "import sys; print(sys.executable)" points inside the environment.

For the later AWS labs, install boto3 with python -m pip install boto3. Record tested dependencies with python -m pip freeze > requirements.txt, and recreate them with python -m pip install -r requirements.txt. Keep .venv/ out of version control. See the Python virtual environment tutorial.

Sample input and task

Turn a list of server states into a command-line report. Save the solution as report.py. Accept a status filter, log a count to stderr, print matching names to stdout, and return zero on success.

Run python report.py --status critical. Expected stdout: web-02. Stderr includes INFO matched=1. An unsupported status must produce an argument error and a nonzero exit code.

import argparse
import logging
import os


def select_names(records, status):
    return [row["name"] for row in records if row["status"] == status]


def main():
    parser = argparse.ArgumentParser(description="Filter server states")
    parser.add_argument("--status", choices=["healthy", "critical"],
                        default="critical")
    args = parser.parse_args()
    level = os.environ.get("REPORT_LOG_LEVEL", "INFO").upper()
    if level not in {"DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"}:
        parser.error("REPORT_LOG_LEVEL must be a logging level")
    logging.basicConfig(level=level, format="%(levelname)s %(message)s")
    records = [{"name": "web-01", "status": "healthy"},
               {"name": "web-02", "status": "critical"}]
    names = select_names(records, args.status)
    logging.info("matched=%d", len(names))
    for name in names:
        print(name)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

Test the behavior

Save this beside the script as test_report.py and run python -m unittest -v.

import unittest
from report import select_names


class ReportTests(unittest.TestCase):
    def test_filters_without_changing_input(self):
        rows = [{"name": "web-01", "status": "healthy"},
                {"name": "web-02", "status": "critical"}]
        self.assertEqual(select_names(rows, "critical"), ["web-02"])
        self.assertEqual(len(rows), 2)

    def test_empty_and_no_match(self):
        self.assertEqual(select_names([], "critical"), [])
        self.assertEqual(select_names(
            [{"name": "web-01", "status": "healthy"}], "critical"), [])

Before adding an API

For boto3, use Config(connect_timeout=5, read_timeout=20, retries={"mode": "standard", "total_max_attempts": 4}). The attempt limit includes the original request. See the SDK retry documentation.

Completion checklist and challenge

• Run both status filters and the invalid-status case.
• Pass the unit tests without calling a network service.
• Explain the difference between logs and report output.
• Add a CSV input argument and a test for a missing required column.

Continue to the EC2 report lab and then the inventory capstone.

Prerequisites

Complete the production scripting bridge. This exercise initially uses only Python. The optional live extension needs boto3, configured AWS credentials, a region, and ec2:DescribeInstances permission.

Sample input

Use this simulated API page; no instance needs to be created:

sample = {"Reservations": [{"Instances": [
    {"InstanceId": "i-example1", "State": {"Name": "running"},
     "InstanceType": "t3.micro", "Tags": [{"Key": "Name", "Value": "web-01"}]},
    {"InstanceId": "i-example2", "State": {"Name": "stopped"},
     "InstanceType": "t3.micro"}
]}]}

Task and expected output

Write rows_from_pages(pages, tag_key=None) to process every page, return normalized rows sorted by state and ID, support a tag-key filter, and use unnamed when the Name tag is absent.

running | i-example1 | web-01 | t3.micro
stopped | i-example2 | unnamed | t3.micro

With tag_key="Name", only the first row remains.

def rows_from_pages(pages, tag_key=None):
    rows = []
    for page in pages:
        for reservation in page.get("Reservations", []):
            for instance in reservation.get("Instances", []):
                tags = {tag["Key"]: tag["Value"]
                        for tag in instance.get("Tags", [])}
                if tag_key is not None and tag_key not in tags:
                    continue
                rows.append({
                    "id": instance["InstanceId"],
                    "state": instance["State"]["Name"],
                    "type": instance["InstanceType"],
                    "name": tags.get("Name", "unnamed"),
                })
    return sorted(rows, key=lambda row: (row["state"], row["id"]))


# Put the sample definition above this call.
for row in rows_from_pages([sample]):
    print(f"{row['state']} | {row['id']} | {row['name']} | {row['type']}")

Check your work

• Empty Reservations returns no rows.
• Splitting the sample across two pages still returns both instances.
• Filtering by an absent tag returns no rows.
• Reversing page order does not change report order.
• Missing tags do not raise an exception.

Optional live extension

Pass client.get_paginator("describe_instances").paginate() to your function. Configure timeouts and bounded SDK retries as shown in the bridge. A single API response may not cover the fleet; see the DescribeInstances paginator.

Live output depends on your account and region. Credential and permission errors are failures, not empty inventories. The capstone includes a complete CLI and error handling.

Challenge

Filter running instances missing an Owner tag. Explain the difference between a missing tag and a present tag with an empty value. Export rows as CSV and test against the local sample.

Prerequisites

Know functions, exceptions, and logging from the production scripting bridge. This simulation uses only Python and makes no network calls.

Sample input and task

A fake service raises ConnectionError twice, then returns ok. Allow four total attempts, wait 0.1 then 0.2 seconds between the first three attempts, and log each failure. Propagate unrelated errors immediately.

Expected output

Warnings indicate attempts 1 and 2 failed, then stdout prints ok. No sleep follows success or the last failure. Four total attempts means one original call and at most three retries.

import logging
import time


def run_with_retry(operation, max_attempts=4, base_delay=0.1, sleep=time.sleep):
    if max_attempts < 1 or base_delay < 0:
        raise ValueError("Invalid retry configuration")
    for attempt in range(1, max_attempts + 1):
        try:
            return operation()
        except ConnectionError:
            if attempt == max_attempts:
                logging.error("attempt=%d exhausted=true", attempt)
                raise
            delay = min(base_delay * 2 ** (attempt - 1), 5.0)
            logging.warning("attempt=%d retry_in=%.2f", attempt, delay)
            sleep(delay)


def main():
    logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
    outcomes = iter([ConnectionError("temporary"), ConnectionError("temporary"), "ok"])

    def fake_call():
        result = next(outcomes)
        if isinstance(result, Exception):
            raise result
        return result

    print(run_with_retry(fake_call))


if __name__ == "__main__":
    main()

Check your work

Replace sleep with delays.append, where delays = [], to test without waiting.

Challenge and production connection

Add bounded jitter and test delay ranges instead of exact values. For a real service, use the client’s exception types and timeouts. Retrying writes requires an idempotency strategy.

For boto3, use SDK retries instead of wrapping them in this simulation: nested policies multiply attempts. The sample logs key/value text; extend it to JSON if your log collector requires JSON events.

Continue to request pacing.

Prerequisites

Complete the retry lab. This simulation uses Python’s standard library in a single process.

Sample input and task

Send 50 simulated requests at no more than 10 starts per second by leaving at least 0.1 seconds between starts. Use a monotonic clock so wall-clock adjustments cannot change the pacing.

Expected output

Print Sent 50 requests. The final start occurs at least 4.9 seconds after the first. Real scheduling may be slower. Do not sleep after the last request.

import math
import time


def paced_calls(send, count, rate, clock=time.monotonic, sleep=time.sleep):
    if count < 0 or not math.isfinite(rate) or rate <= 0:
        raise ValueError("Count must be nonnegative and rate positive and finite")
    interval = 1.0 / rate
    next_start = clock()
    for number in range(count):
        while True:
            delay = next_start - clock()
            if delay <= 0:
                break
            sleep(delay)
        next_start = clock() + interval
        send(number)


if __name__ == "__main__":
    sent = []
    paced_calls(sent.append, count=50, rate=10)
    print(f"Sent {len(sent)} requests")

Check your work

Use a fake clock to avoid waiting during tests:

now = [0.0]
starts = []


def clock():
    return now[0]


def sleep(seconds):
    now[0] += seconds


paced_calls(lambda number: starts.append(clock()), 50, 10, clock, sleep)
assert len(starts) == 50
assert starts[-1] >= 4.9 - 1e-9
assert all(b - a >= 0.1 - 1e-9 for a, b in zip(starts, starts[1:]))

Also test zero requests, invalid rates, and a send function that advances time by 0.3 seconds. Slow calls must not cause a catch-up burst.

Knowledge check and challenge

Does this enforce a shared limit across workers? No. Each process has its own schedule. Shared quotas need coordination.

Extend the simulation with a rate-limit response that specifies a retry delay. Respect that delay. Real APIs can use fixed windows, sliding windows, or token buckets; validate the service policy instead of treating this example as a universal limiter.

Continue to the inventory capstone.

Prerequisites

Complete the production scripting bridge, EC2 reporting, and retry labs. The offline project needs Python 3 and no third-party packages. Optional live mode needs boto3, configured credentials, and ec2:DescribeInstances permission in your chosen region.

Project brief

An operations team needs a repeatable EC2 inventory report. Build a command-line tool that:

1. Reads either local response fixtures or all EC2 response pages in one region.
2. Produces a deterministic CSV with region, ID, state, type, and name.
3. Handles missing Name tags and empty inventories.
4. Uses connection/read timeouts and at most four SDK attempts per request.
5. Logs a count and region, and returns a nonzero exit code on failure.
6. Refuses to overwrite an existing report and removes a partial report if writing fails.
7. Can be tested without credentials or network calls.

Sample input and downloads

Save these files in the same directory. The IDs in the fixture are fictional.

• sample-pages.json — two pages with one instance each.
• test_inventory.py — behavior tests for your implementation.

Write your own inventory.py before opening the reference implementation below.

Expected output

python inventory.py --fixture sample-pages.json --region us-east-1 --output inventory.csv

Stdout: Wrote 2 instances to inventory.csv. The file contains:

region,id,state,type,name
us-east-1,i-example1,running,t3.micro,web-01
us-east-1,i-example2,stopped,t3.micro,unnamed

The region flag labels offline fixture records; fixtures themselves contain no region metadata. A second run using the same output path must fail without changing the first report. Use a new path for the next successful run.

Run the tests

python -m unittest -v test_inventory.py

Tests cover multiple pages, missing tags, stable ordering, empty output, malformed input, output preservation, partial-write cleanup, and propagated pagination errors. A mocked paginator exercises the collection contract; it does not simulate the SDK retry engine.

Optional live run

Install boto3 in your virtual environment and use your configured AWS profile or role:

python -m pip install boto3
python inventory.py --live --region us-east-1 --output live-inventory.csv

Add --profile training if you use a named profile. This mode reads EC2 inventory and does not create or modify AWS resources. Permission or credential failures must fail the run; they must not become a successful empty report.

The SDK handles supported transient errors within the configured attempt budget. See SDK retries and the EC2 paginator. A whole-job deadline, distributed rate limits, and multi-account credentials are outside this first version.

Completion checklist

• Generate the exact sample report and pass the offline tests.
• Explain why only reading the first response page loses data.
• Demonstrate a failed run leaves no new report.
• Explain timeouts, total attempts, and failure exit codes.
• Give a five-minute walkthrough of your design and one limitation.

Further challenges

Add multiple regions, a missing-Owner-tag report, and a scheduled CI job that publishes the CSV as an artifact. Test one failed region and decide whether partial results should count as success. If users open reports in spreadsheet software, add and test a policy for untrusted tag values that resemble formulas.

Continue to production scenarios or the interview revision route.