Python Error Tracker with CloudWatch

Build a Python error-handling pipeline that ships structured logs to AWS CloudWatch.

Introduction

30 Second Summary

Every application breaks eventually. The difference between a frustrating mystery and a quick fix comes down to how well you track and categorize your errors.

In this project, you will build a Python order processing system that fails in realistic ways, catches every error with structured logging, and ships those logs to AWS CloudWatch for searching and analysis.

What You'll Build

Run a simulated order processor, watch it handle payments and inventory checks with a mix of successes and categorized failures, then jump into the CloudWatch console to query exactly which errors happened and when.

By the end of this project, you'll have:

  • A running order processor that handles 20 orders with realistic failures like payment timeouts, out-of-stock items, and shipping errors.
  • Structured JSON logs flowing from your local machine to CloudWatch, persisting in the cloud and searchable from any browser.
  • Hands-on experience querying your logs with CloudWatch Logs Insights to answer questions like "which error type is most common?" and "when did errors spike?".
  • Secret Mission: Set up a CloudWatch Alarm that emails you when errors spike above a threshold.

Are there any prerequisites?

You need an AWS account (free tier eligible) to complete this project. Step 1 walks you through installing Python, the AWS CLI, and all required libraries from scratch.

Before We Start

To build an error tracker that ships logs to the cloud, you need three things in place: Python to write the app, the AWS CLI to talk to Amazon's cloud services, and the right permissions for your code to create log entries in CloudWatch.

This step gets your entire development environment ready. By the end, you will have Python, the AWS CLI, and all project libraries installed and verified.

In this step, get ready to:
  • Install Python 3.10+ and the AWS CLI v2.
  • Configure your AWS credentials and grant CloudWatch Logs permissions.
  • Create a project folder with a virtual environment and install dependencies.
Install Python and the AWS CLI

Python runs your application code. The AWS CLI lets you interact with AWS services from your terminal and stores the credentials that your Python code will use automatically.

  • Open your terminal. Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Terminal (macOS) or PowerShell (Windows) and press Enter to open it.
  • Check if Python 3.10+ is already installed by running:
python3 --version

✔️ I see version 3.10 or higher

Python is ready. Move on to the AWS CLI check below.

ⓧ I see a version lower than 3.10

You need Python 3.10 or higher because boto3 (the AWS SDK for Python) dropped support for Python 3.9 in April 2026.

  • Download the latest Python from https://www.python.org/downloads/.
  • Run the installer. On macOS, the installer places the new binary at a versioned path. On Windows, check Add python.exe to PATH during install.
  • Close and reopen your terminal, then run python3 --version again to confirm 3.10 or higher appears.

ⓧ Command not found

  • Download the latest Python from https://www.python.org/downloads/.
  • Run the installer. On macOS, follow the prompts. On Windows, check Add python.exe to PATH during install.
  • Close and reopen your terminal, then run python3 --version to confirm 3.10 or higher appears.
  • Now install the AWS CLI v2. Follow the tab for your operating system:

macOS

  • Download and install the official AWS CLI package by running these commands:
curl "https://awscli.amazonaws.com/AWSCLIV2.pkg" -o "AWSCLIV2.pkg"
sudo installer -pkg AWSCLIV2.pkg -target /

What do these commands do?

The first command downloads the official installer package from AWS. The second command runs the macOS installer, which places the aws binary at /usr/local/bin/aws.

Your terminal will prompt you for your system password because sudo grants write permission to system folders.

Windows

  • Download and run the official AWS CLI MSI installer by running this command in PowerShell:
msiexec.exe /i https://awscli.amazonaws.com/AWSCLIV2.msi

What does this command do?

This launches the MSI installer for the AWS CLI v2. Follow the on-screen prompts to complete the installation. You may need admin rights.

  • Close and reopen your terminal after the installer finishes.
  • Confirm the AWS CLI is installed by running:
aws --version

You should see output starting with aws-cli/2. followed by a version number.

Command not found after installing?

  • Close your terminal completely and reopen it. The PATH update takes effect in a new session.
  • On macOS, verify the symlink exists by running which aws. It should show /usr/local/bin/aws.
  • On Windows, search for cmd in the Start menu and try aws --version there instead.

Still stuck? Help me troubleshoot my AWS CLI installation.

Configure AWS credentials and permissions

The AWS CLI needs to know which AWS account to send logs to. You will provide it with access keys from your IAM user, then grant that user permission to write to CloudWatch Logs.

  • Configure the AWS CLI by running:
aws configure

What will aws configure ask for?

The command prompts you for four values. Enter them one at a time:

  • AWS Access Key ID and AWS Secret Access Key: find these in the AWS Console under IAM > Users > your user > Security credentials > Create access key.
  • Default region name: enter us-east-1.
  • Default output format: enter json.
  • Now grant your IAM user permission to write CloudWatch Logs. Open the AWS Console in your browser.
  • Navigate to the IAM service.
  • Click Users in the left sidebar, then click your username.
  • Click Add permissions, then select Attach policies directly.
  • Search for CloudWatchLogsFullAccess, check the box next to it, and click Next.
  • Click Add permissions to confirm.

What is CloudWatchLogsFullAccess?

This is an AWS managed policy that grants full access to CloudWatch Logs actions. It allows your code to create log groups, create log streams, and send log events. In production you would scope permissions more tightly, but for learning it keeps things simple.

  • Before you run the next command, do you think it will succeed or fail given what you just configured?
  • Verify your credentials are working by running:
aws sts get-caller-identity

You should see a JSON response containing your AWS account ID, user ARN, and user ID. This confirms the CLI can authenticate against your account.

Seeing an authentication error?

  • Run aws configure again and double-check that you pasted your Access Key ID and Secret Access Key without extra spaces.
  • Make sure the IAM user whose keys you used is active (not disabled) in the IAM console.
  • Confirm you chose us-east-1 as the region.

Still stuck? Help me fix my AWS CLI authentication error.

Create the project and install dependencies

Your code needs a dedicated folder and an isolated Python environment so its packages do not conflict with anything else on your machine.

  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir error-tracker
cd error-tracker
  • Create and activate a Python virtual environment:

macOS

python3 -m venv venv
source venv/bin/activate

Windows

python3 -m venv venv
venv\Scripts\activate

Your terminal prompt now shows (venv) at the beginning. This confirms the virtual environment is active.

  • Create a requirements.txt file to declare your project dependencies. In VS Code, open the error-tracker folder by clicking File > Open Folder and selecting it from your Desktop.
  • Create a new file called requirements.txt and add the following content:
watchtower==3.4.0
boto3

What are these packages?

  • watchtower is a Python logging handler that ships log events directly to AWS CloudWatch Logs. It handles batching, sequencing, and delivery for you.
  • boto3 is the official AWS SDK for Python. It provides the underlying connection that watchtower uses to reach CloudWatch.
  • Install the dependencies by running:
pip install watchtower==3.4.0 boto3
  • Verify both packages imported correctly by running:
python3 -c "import watchtower; import boto3; print('Ready')"

You should see Ready printed to the terminal. This confirms both libraries are installed and importable.

Seeing a ModuleNotFoundError?

  • Make sure your virtual environment is active. You should see (venv) in your terminal prompt.
  • If you accidentally installed packages outside the venv, deactivate, delete the venv folder, recreate it, activate again, and rerun pip install.
  • On some systems, use pip3 instead of pip if the command is not found.

Still stuck? Help me fix my Python package import error.

✔️ Awesome, I've got everything!

Great. Double check you saved requirements.txt and that your terminal still shows (venv) as active.

ⓧ I'd like to double check the full code

Here is the complete requirements.txt for comparison:

watchtower==3.4.0
boto3

Your development environment is locked and loaded. Next up, you will build the order processing application that generates realistic errors for your logging pipeline to capture.

Build the Order Processor with print()

Your environment is ready with Python, the AWS CLI, and all the packages you need. Now you need an application that actually produces errors worth tracking.

In this step, you will build a simulated order processor that handles payments, inventory checks, and shipping. Some orders will fail in realistic ways, and you will catch those failures with print() statements. By the end, you will see firsthand why print() falls short once an application grows.

In this step, get ready to:
  • Create custom exception classes for payment, inventory, and shipping failures.
  • Build processing functions that randomly trigger those failures.
  • Run the app and observe why print() output becomes hard to work with.
Create the exception classes

Every order can fail for a different reason: a payment gets declined, an item is out of stock, or shipping cannot reach a region. Custom exception classes let you catch each failure type separately.

  • Create a new file called order_processor.py in your project folder by running:
touch order_processor.py
  • Open order_processor.py in VS Code and paste the following code:
import random
import time


class OrderError(Exception):
    """Base exception for order processing errors."""
    pass


class PaymentError(OrderError):
    """Raised when payment processing fails."""
    pass


class InventoryError(OrderError):
    """Raised when an item is out of stock."""
    pass


class ShippingError(OrderError):
    """Raised when shipping calculation fails."""
    pass

What does this code do?

OrderError is a base exception that all your custom errors inherit from. This lets you catch ALL order-related errors with a single except OrderError if you ever need to.

PaymentError, InventoryError, and ShippingError each represent a specific failure category. This hierarchy is a standard Python pattern for organizing exceptions by domain.

  • Save order_processor.py.

Now add the functions that simulate real processing steps and raise these exceptions.

  • Add the following three functions below the ShippingError class in order_processor.py:
def validate_payment(order):
    """Simulate payment validation."""
    if order["amount"] > 1000:
        raise PaymentError(
            f"Payment declined: amount ${order['amount']} exceeds limit"
        )
    if random.random() < 0.15:
        raise PaymentError("Payment gateway timeout")


def check_inventory(order):
    """Simulate inventory check."""
    out_of_stock_items = ["ITEM-999", "ITEM-500"]
    if order["item_id"] in out_of_stock_items:
        raise InventoryError(f"Item {order['item_id']} is out of stock")
    if random.random() < 0.1:
        raise InventoryError("Inventory service unavailable")


def calculate_shipping(order):
    """Simulate shipping calculation."""
    invalid_regions = ["REGION-X", "REGION-Z"]
    if order.get("region") in invalid_regions:
        raise ShippingError(f"Cannot ship to region: {order['region']}")
    if random.random() < 0.05:
        raise ShippingError("Shipping rate API error")

What does this code do?

Each function simulates a real-world validation step that can fail:

  • validate_payment rejects orders over $1000 and randomly times out 15% of the time.
  • check_inventory flags specific items as out of stock and randomly fails 10% of the time.
  • calculate_shipping blocks certain regions and randomly errors 5% of the time.
  • Save order_processor.py.
  • Confirm the module loads without errors by running:
python3 -c "from order_processor import validate_payment; print('Module loads OK')"

You should see Module loads OK printed in your terminal.

Seeing an ImportError?

  • Make sure you saved order_processor.py before running the command.
  • Check that your terminal is in the same directory as the file. Run ls to confirm order_processor.py is listed.
  • Make sure your virtual environment is still active (you should see (venv) in your prompt).

Help me debug this ImportError in my Python project.

Add order generation and the main loop

The exception classes and validation functions define what can go wrong. Now you need a function that orchestrates the full order lifecycle and catches each error type.

  • Add the following process_order function below calculate_shipping in order_processor.py:
def process_order(order):
    """Process a single order through all validation steps."""
    order_id = order["order_id"]
    print(f"Processing order {order_id}")
    try:
        validate_payment(order)
        check_inventory(order)
        calculate_shipping(order)
        print(f"Order {order_id} completed successfully")
        return True
    except PaymentError as e:
        print(f"Payment failed for order {order_id}: {e}")
        return False
    except InventoryError as e:
        print(f"Inventory check failed for order {order_id}: {e}")
        return False
    except ShippingError as e:
        print(f"Shipping failed for order {order_id}: {e}")
        return False
    except Exception as e:
        print(f"Unexpected error processing order {order_id}: {e}")
        return False

What does this code do?

process_order runs one order through all three validation steps in sequence. If any step raises an exception, the matching except block catches it and prints a failure message.

The final except Exception is a safety net for any unexpected error that does not match the custom types. Every path returns True or False so the caller can tally results.

  • Save order_processor.py.
  • Test the function with a single order by running:
python3 -c "from order_processor import process_order; process_order({'order_id': 'TEST-001', 'amount': 50, 'item_id': 'ITEM-001', 'region': 'US-EAST'})"

You should see Processing order TEST-001 followed by either a success or failure message.

Seeing a NameError?

  • Check that process_order is defined below the exception classes and below the three validation functions. Python reads files top-to-bottom, so any function you call must be defined above the caller.
  • Confirm your indentation is consistent (4 spaces per level, no tabs mixed in).

Help me fix a NameError in my order processor.

Now add the test data generator and the main entry point that processes a full batch of 20 orders.

  • Add the following code below process_order in order_processor.py:
def generate_test_orders(count=20):
    """Generate a batch of test orders with realistic variation."""
    items = ["ITEM-001", "ITEM-002", "ITEM-003", "ITEM-999", "ITEM-500", "ITEM-100"]
    regions = ["US-EAST", "US-WEST", "EU-WEST", "REGION-X", "REGION-Z", "AP-SOUTH"]
    orders = []
    for i in range(count):
        orders.append({
            "order_id": f"ORD-{i + 1:04d}",
            "item_id": random.choice(items),
            "amount": random.choice([49.99, 99.99, 250.00, 500.00, 1500.00]),
            "region": random.choice(regions),
        })
    return orders


def main():
    orders = generate_test_orders(20)
    results = {"success": 0, "failed": 0}
    for order in orders:
        success = process_order(order)
        if success:
            results["success"] += 1
        else:
            results["failed"] += 1
        time.sleep(0.1)
    print(f"\nBatch Summary: {results['success']} succeeded, {results['failed']} failed")


if __name__ == "__main__":
    main()

What does this code do?

  • generate_test_orders creates 20 orders with randomized items, amounts, and regions. Some combinations are designed to trigger failures (like ITEM-999 or amounts over $1000).
  • main() runs the full batch, tallies successes and failures, and prints a summary at the end.
  • time.sleep(0.1) adds a small delay between orders so you can watch the output stream in real time.
  • Save order_processor.py.

✔️ Awesome, I've got everything!

Great. Make sure you have saved order_processor.py with all four code chunks in order.

ⓧ I'd like to double check the full code

Scroll up through your file and confirm these sections appear in this exact order:

  • Imports (random and time).
  • Four exception classes (OrderError, PaymentError, InventoryError, ShippingError).
  • Three validation functions (validate_payment, check_inventory, calculate_shipping).
  • The process_order function with try/except blocks and print() calls.
  • The generate_test_orders function.
  • The main() function and the if __name__ == "__main__" block at the very end.
Run the app and observe the output

Your order processor is complete. Before you run it, think about what you expect to see in the terminal.

  • Predict: do you think you will be able to tell errors from successes at a glance in the output?
  • Run the order processor by executing:
python3 order_processor.py

You should see 20 orders process with a mix of success and failure messages, followed by a batch summary.

What should I see?

Your terminal fills with lines like:

Processing order ORD-0001
Order ORD-0001 completed successfully
Processing order ORD-0002
Payment failed for order ORD-0002: Payment gateway timeout
Processing order ORD-0003
Inventory check failed for order ORD-0003: Item ITEM-999 is out of stock
...
Batch Summary: 12 succeeded, 8 failed

Your exact numbers will differ because failures are randomized.

Now look closely at the output. Every line looks the same.

Why is this output a problem?

You cannot tell errors from successes at a glance because there is no severity label. There are no timestamps, so you cannot tell when anything happened.

If you close the terminal, the output is gone forever. You cannot filter, search, or analyze this output without manually reading every line. In the next step, Python's logging module fixes all three problems.

App not running?

  • If you see SyntaxError, check for mismatched quotes or missing colons on your class and def lines.
  • If you see ModuleNotFoundError: No module named 'order_processor', make sure you are running the command from the same folder where order_processor.py lives.
  • If nothing happens at all, confirm the if __name__ == "__main__" block is at the bottom of your file with no extra indentation.

Help me debug why my order processor script won't run.

Your order processor works, but the output is a wall of undifferentiated text. Next up, you will replace those print() calls with Python's logging module and JSON formatting, giving every line a timestamp, severity level, and structured fields you can search.

Replace print() with Structured Logging

In the previous step, you ran your order processor and saw output in the terminal. Every message looked the same. Successes and failures were indistinguishable at a glance.

Python's built-in logging module solves three problems that print() cannot: it tags every message with a severity level (INFO, ERROR, DEBUG), it adds timestamps automatically, and it outputs structured data you can filter and search. In this step, you'll build a custom JSON formatter and replace every print() call with a proper logger call.

In this step, get ready to:
  • Create a custom JSON formatter that outputs structured log records.
  • Replace all print() calls with leveled logger calls (info, error, debug, critical).
  • Run the app and see JSON-formatted output with timestamps and severity levels.
Create the JsonFormatter class

Python's logging module uses a three-part architecture: a logger receives messages, a handler decides where they go, and a formatter controls what they look like. You'll build a custom formatter that outputs JSON.

  • In order_processor.py, find the two import lines at the top of the file:
import random
import time
  • Replace those two lines with these expanded imports:
import logging
import json
import random
import time
from datetime import datetime

What do these new imports do?

  • logging is Python's built-in module for structured log output with severity levels.
  • json lets you serialize log data into JSON strings that tools can parse.
  • datetime provides the UTC timestamp you'll attach to every log record.
  • Below the ShippingError class (after the line pass), add the JsonFormatter class by pasting this code:
class JsonFormatter(logging.Formatter):
    """Formats log records as JSON for structured logging."""

    def format(self, record):
        log_data = {
            "timestamp": datetime.utcnow().isoformat() + "Z",
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
        }
        if record.exc_info:
            log_data["exception"] = self.formatException(record.exc_info)
        if hasattr(record, "order_id"):
            log_data["order_id"] = record.order_id
        if hasattr(record, "error_type"):
            log_data["error_type"] = record.error_type
        return json.dumps(log_data)

What does this class do?

JsonFormatter extends Python's logging.Formatter and overrides the format() method. Instead of producing a plain text string, it builds a Python dictionary with structured fields and serializes it to JSON.

The extra fields (order_id, error_type) are only included when present. This means your logs automatically carry context about which order failed and what category of error it was.

  • Directly below the JsonFormatter class (and above the validate_payment function), add the setup_logging() function:
def setup_logging():
    """Configure logging with a console handler."""
    logger = logging.getLogger("order_processor")
    logger.setLevel(logging.DEBUG)

    console_handler = logging.StreamHandler()
    console_handler.setLevel(logging.INFO)
    console_handler.setFormatter(JsonFormatter())

    logger.addHandler(console_handler)

    return logger

What does setup_logging do?

  • getLogger("order_processor") creates a named logger you can retrieve from anywhere in your app.
  • setLevel(logging.DEBUG) on the logger means it accepts all severity levels. The handler then filters what actually gets printed.
  • The StreamHandler prints to the terminal. Its level is set to INFO, so DEBUG messages are captured but not shown in the console output yet.
  • Save order_processor.py.
  • Test that the formatter works by running this quick check:
python3 -c "from order_processor import setup_logging; logger = setup_logging(); logger.info('test')"

You should see a single JSON line printed to your terminal, something like:

{"timestamp": "2026-07-22T10:30:00.123456Z", "level": "INFO", "logger": "order_processor", "message": "test"}

Don't see JSON output?

  • Make sure you saved order_processor.py before running the command.
  • Check that your virtual environment is still active (you should see (venv) in your terminal prompt).
  • Verify that the JsonFormatter class is placed after ShippingError and before validate_payment.

Still stuck? Why isn't my JSON formatter producing output?

Replace print() with logger calls

Now that your logging infrastructure works, you'll swap every print() call in process_order and main with proper logger methods. Each method corresponds to a severity level: info for normal operations, error for caught exceptions, debug for granular progress, and critical for unexpected failures.

  • In order_processor.py, find the existing process_order function. Delete it entirely (from def process_order(order): down to the last return False).
  • Replace it with this updated version. First, add the function definition and the success path inside the try block:
def process_order(order, logger):
    """Process a single order through all validation steps."""
    order_id = order["order_id"]

    logger.info(
        f"Processing order {order_id}",
        extra={"order_id": order_id},
    )

    try:
        validate_payment(order)
        logger.debug(
            f"Payment validated for order {order_id}",
            extra={"order_id": order_id},
        )

        check_inventory(order)
        logger.debug(
            f"Inventory confirmed for order {order_id}",
            extra={"order_id": order_id},
        )

        calculate_shipping(order)
        logger.debug(
            f"Shipping calculated for order {order_id}",
            extra={"order_id": order_id},
        )

        logger.info(
            f"Order {order_id} completed successfully",
            extra={"order_id": order_id},
        )
        return True
  • Immediately after that code (still inside the same function), add the except blocks that handle each error type:
    except PaymentError as e:
        logger.error(
            f"Payment failed for order {order_id}: {e}",
            extra={"order_id": order_id, "error_type": "PaymentError"},
        )
        return False

    except InventoryError as e:
        logger.error(
            f"Inventory check failed for order {order_id}: {e}",
            extra={"order_id": order_id, "error_type": "InventoryError"},
        )
        return False

    except ShippingError as e:
        logger.error(
            f"Shipping failed for order {order_id}: {e}",
            extra={"order_id": order_id, "error_type": "ShippingError"},
        )
        return False

    except Exception as e:
        logger.critical(
            f"Unexpected error processing order {order_id}: {e}",
            extra={"order_id": order_id, "error_type": "UnexpectedError"},
            exc_info=True,
        )
        return False

What do the log levels mean?

  • logger.info() marks routine operations: order started, order completed.
  • logger.debug() marks granular progress: payment validated, inventory confirmed. These are captured but hidden from the console (the handler level is INFO).
  • logger.error() marks known failure categories with the error_type extra field so you can filter by category later.
  • logger.critical() with exc_info=True captures the full stack trace for truly unexpected failures.
  • Now find the existing main() function (below generate_test_orders). Delete it entirely and replace it with this updated version:
def main():
    logger = setup_logging()

    logger.info("Order processing batch started")

    orders = generate_test_orders(20)
    results = {"success": 0, "failed": 0}

    for order in orders:
        success = process_order(order, logger)
        if success:
            results["success"] += 1
        else:
            results["failed"] += 1
        time.sleep(0.1)

    logger.info(
        f"Batch complete: {results['success']} succeeded, {results['failed']} failed"
    )

    print(
        f"\nBatch Summary: {results['success']} succeeded, {results['failed']} failed"
    )


if __name__ == "__main__":
    main()

What changed in main()?

The function now calls setup_logging() to create the logger, then passes it to each process_order() call. The batch start and completion are logged with logger.info(). A final print() remains for a quick human-readable summary at the bottom.

  • Save order_processor.py.

Seeing a syntax error?

  • Make sure the except blocks are indented to align with the try: statement inside process_order. Each except should be at 4 spaces of indentation.
  • Check that you deleted the old process_order function completely before pasting the new one. Duplicate function definitions cause errors.
  • Verify that if __name__ == "__main__": appears at the very bottom of the file with no indentation.

Still stuck? Help me fix a syntax error in my order_processor.py after replacing print with logger calls.

Run the app and compare the output

Before you run, think about what you expect to see. The app is processing the same 20 orders with the same random failures as before.

  • What do you think the terminal output will look like now compared to Step 2? Will it be easier or harder to spot the errors?
  • Run the updated app:
python3 order_processor.py

Every line is now a JSON object. You can see the "level": "INFO" or "level": "ERROR" field on each line, along with a timestamp and the logger name. Error lines also carry an error_type field telling you exactly which category of failure occurred.

Why is this better than print()?

With print(), every message was an undifferentiated string. Now you could pipe this output to a file and run grep ERROR to see only failures. You could parse the JSON to count errors by type. The structure makes your logs machine-readable without losing human readability.

✔️ Awesome, I've got everything!

Great. Double check you've saved order_processor.py.

ⓧ I'd like to double check the full code

Here is the complete order_processor.py at this point. Your file is long, so confirm these sections are present in order:

  • Imports (logging, json, random, time, datetime) at the top.
  • Exception classes (OrderError, PaymentError, InventoryError, ShippingError).
  • JsonFormatter class.
  • setup_logging() function.
  • validate_payment, check_inventory, calculate_shipping functions (unchanged).
  • process_order(order, logger) with logger calls and extra fields.
  • generate_test_orders() (unchanged).
  • main() calling setup_logging() and passing logger to process_order.
  • if __name__ == "__main__": at the very bottom.

Scroll through your file and confirm each section is present, in order, with no duplicates.

Your app now produces structured, leveled, timestamped logs. Next up, you'll send these logs to AWS CloudWatch so they persist in the cloud and survive your terminal session.

Send Logs to CloudWatch

Your app now produces clean, structured JSON logs in the terminal. But here is the problem: close that terminal window and every log line vanishes forever.

In this step, you will connect your application to AWS CloudWatch, a centralized log management service. Once connected, your logs persist in the cloud, searchable from anywhere, across every run.

In this step, get ready to:
  • Add a CloudWatch log handler to your Python logger using the watchtower library.
  • Run your app and verify that logs appear in the AWS CloudWatch console.
  • Prove that logs persist across multiple runs by running the app a second time.
Add the CloudWatch log handler

The watchtower library provides a Python logging handler that sends log events directly to CloudWatch. It uses your existing AWS CLI credentials automatically via boto3.

  • In order_processor.py, find the import line from datetime import datetime at the top of the file.
  • Add these two import lines directly below it:
import watchtower
import boto3
  • Now scroll down to the setup_logging() function. Find the line logger.addHandler(console_handler).
  • Add the following CloudWatch handler code between logger.addHandler(console_handler) and return logger:
    # Send logs to AWS CloudWatch for persistent, centralized storage
    cloudwatch_handler = watchtower.CloudWatchLogHandler(
        log_group_name="/learning/order-processor",
        log_stream_name="orders-{:%Y-%m-%d}".format(datetime.now()),
        boto3_client=boto3.client("logs", region_name="us-east-1"),
    )
    cloudwatch_handler.setLevel(logging.DEBUG)
    cloudwatch_handler.setFormatter(JsonFormatter())

    logger.addHandler(cloudwatch_handler)

What does this code do?

This creates a second logging handler that ships every log event to CloudWatch in addition to printing it to your terminal.

  • log_group_name sets the CloudWatch Log Group, which is like a folder for related logs.
  • log_stream_name creates a date-stamped stream within the group, so each day gets its own stream.
  • boto3_client tells watchtower which AWS region to use. It picks up your credentials from the AWS CLI configuration you set up earlier.
  • Also update the setup_logging() docstring from "Configure logging with a console handler." to "Configure logging with console and CloudWatch handlers.".
Add a flush to ensure log delivery

The watchtower handler batches log events before sending them. You need to flush the handlers at the end of your script to make sure all logs are delivered before the program exits.

  • In the main() function, find the logger.info call that logs the batch complete message.
  • Add this flush loop and updated print statements between the logger.info(...) call and the existing print( line:
    # Flush all handlers to ensure CloudWatch receives every log event
    for handler in logger.handlers:
        handler.flush()
  • Finally, add this line after the existing print( statement at the bottom of main():
    print("Check CloudWatch Logs console for full log history.")
  • Save order_processor.py.

Seeing a syntax error when you save?

Make sure the indentation of the new code matches the surrounding lines. The handler block and flush loop should be indented with 4 spaces, matching the existing code inside each function.

Help me fix my indentation error

Run your app and find your logs in CloudWatch

Before you run this, think about what you expect to happen. Will the terminal output change? Where else should your logs appear now?

  • Run your order processor by executing this command:
python3 order_processor.py

You should see the same JSON log lines in your terminal as before. The console handler still works. But now those same events were also shipped to CloudWatch in the background.

At the end of the output, you should see the new message: Check CloudWatch Logs console for full log history..

Seeing a NoCredentialsError or AccessDenied?

  • Run aws sts get-caller-identity to confirm your credentials are still configured. If it fails, re-run aws configure.
  • Confirm your IAM user has the CloudWatchLogsFullAccess policy attached (you did this in Step 1).
  • Make sure your virtual environment is activated. You should see (venv) in your terminal prompt.

Help me fix my AWS credentials error

  • Open the AWS CloudWatch console at https://console.aws.amazon.com/cloudwatch/.
  • In the left navigation pane, click Logs.
  • Click Log groups.
  • Find and click the log group named /learning/order-processor.
  • Click the log stream that starts with orders- followed by today's date.

You should see your JSON log events listed. Each event contains the same structured fields (timestamp, level, logger, message) that you saw in the terminal.

Why did the log group appear automatically?

The watchtower handler creates the log group and log stream automatically if they do not already exist. You did not need to pre-create anything in the AWS console.

Prove logs persist across multiple runs

The real power of centralized logging is persistence. Terminal output disappears when you scroll or close the window. CloudWatch keeps everything.

  • Run your order processor a second time:
python3 order_processor.py
  • Return to the CloudWatch console in your browser.
  • Click the log stream name again (or refresh the page).
  • Scroll through the events. You should see log entries from both runs, one after the other.

Both runs are preserved. Even if you close your terminal right now, these logs remain accessible from any browser, on any device.

Why centralized logging matters

In a production environment, your application might run on multiple servers, in containers that get destroyed and recreated, or as serverless functions with no persistent storage. Without centralized logging, those logs are gone the moment the process ends.

CloudWatch gives you a single place to search across all runs, all instances, and all time periods.

✔️ Awesome, I've got everything!

Great. Confirm you have saved order_processor.py and that both runs are visible in CloudWatch.

ⓧ I'd like to double check the full code

Your order_processor.py is over 30 lines, so confirm each section is in place by scrolling through your file:

  • Imports: import watchtower and import boto3 appear after from datetime import datetime.
  • setup_logging(): contains both console_handler and cloudwatch_handler, both added to the logger.
  • cloudwatch_handler uses log_group_name="/learning/order-processor" and log_stream_name="orders-{:%Y-%m-%d}".format(datetime.now()).
  • main(): the for handler in logger.handlers: handler.flush() loop appears after the final logger.info(...) call.
  • main(): ends with both print( lines (the batch summary and the CloudWatch console reminder).

Your logs are flowing to the cloud. Next up, you will learn how to query and analyze those logs using CloudWatch Logs Insights to answer real production questions like "which error type happens most?"

Query Errors with CloudWatch Logs Insights

Your logs are now flowing to CloudWatch and persisting across terminal sessions. That solves the storage problem, but creates a new one: how do you find anything in a growing pile of log events?

Scrolling through raw events line by line does not scale. CloudWatch Logs Insights is a query engine that lets you filter, aggregate, and visualize your logs to answer real questions about your application's health.

In this step, get ready to:
  • Run the default Logs Insights query to view your most recent log events.
  • Write a query that counts errors by type and reveals which failure dominates.
  • Visualize error frequency over time with a bar chart.
Run the default Logs Insights query

Logs Insights gives you a SQL-like query language purpose-built for searching log data. The first step is navigating to the query editor and confirming your logs are reachable.

  • In the CloudWatch console, expand Logs in the left navigation pane.
  • Click Logs Insights.
  • In the Select log group(s) dropdown, search for and select /learning/order-processor.

The query editor already contains a default query:

fields @timestamp, @message | sort @timestamp desc | limit 20
  • Click Run query.

What does this query do?

  • fields @timestamp, @message selects which columns to display for each log event.
  • sort @timestamp desc orders the results with the newest events first.
  • limit 20 caps the output at 20 rows so results load quickly.

You should see a table of your 20 most recent log events, each showing a timestamp and the JSON message body you formatted in the previous step.

Not seeing any results?

  • Make sure you selected the correct log group: /learning/order-processor.
  • Check that the time range in the top-right corner covers the period when you ran your script. Expand it to the last 3 hours if needed.
  • If the log group does not appear in the dropdown, return to your terminal and run the script once more to ensure logs were delivered.

Help me figure out why my Logs Insights query returns no results.

Count errors by type

The default query shows raw events, but cannot answer analytical questions. Now you will write a query that filters for errors, extracts the error_type field from your JSON messages, and counts how often each type occurs.

  • Delete the current query in the editor.
  • Paste the following query:
filter @message like /ERROR/
| parse @message '"error_type": "*"' as error_type
| stats count(*) as error_count by error_type
| sort error_count desc

What does this query do?

  • filter @message like /ERROR/ keeps only log events whose message contains the string ERROR.
  • parse uses a glob pattern to extract the value of error_type from your JSON message into a named field.
  • stats count(*) as error_count by error_type groups the filtered events by error type and counts each group.
  • sort error_count desc puts the most frequent error type at the top of the results.

Before you run this, think: which error type do you expect to appear most often, given how your code is structured?

  • Click Run query.

You should see a table with two columns: error_type and error_count. The most common failure sits at the top of the list.

Why is PaymentError often the most common?

Your code raises a PaymentError when the amount exceeds 1000 (one of five possible amounts is 1500) AND randomly 15% of the time for gateway timeouts. Combined with InventoryError from two out-of-stock items, payment failures tend to dominate.

Your exact numbers will vary because the orders use random data.

Seeing empty results or no error_type column?

  • Check that your query uses straight double quotes inside the parse pattern, not curly quotes that some editors insert.
  • Ensure the time range covers a period when you ran the app. Expand it if needed.
  • Run the app once more in your terminal to generate fresh error data, then re-run the query.

Help me debug my Logs Insights parse query that returns no error_type values.

Visualize error frequency over time

Counting totals answers "what fails most" but not "when did failures cluster together?" A time-series query groups errors into time buckets, which Logs Insights can render as a chart.

  • Delete the current query in the editor.
  • Paste the following time-series query:
filter @message like /ERROR/
| stats count(*) as errors by bin(5m)
  • Click Run query.

What does bin(5m) do?

The bin(5m) function groups your log events into 5-minute time buckets. Instead of one total count, you get a count per bucket, showing how errors are distributed over time.

  • Click the Visualization tab above the results table.
  • If the chart shows a line graph, click the arrow next to Line at the upper left of the graph and select Bar.

You should see a bar chart showing error frequency per 5-minute window. If you only see one or two bars, that is because all your runs happened in a short window.

To generate more data points across multiple time buckets, run your app a few more times from your terminal.

  • Switch back to your terminal and run the app by executing:
python3 order_processor.py
  • Wait about 30 seconds, then run it again:
python3 order_processor.py
  • Return to the Logs Insights tab in your browser.
  • Click Run query again to refresh the visualization with your new data.

Your bar chart should now show multiple bars representing different time buckets, giving you a visual picture of when errors occurred.

Visualization tab not showing a chart?

  • The Visualization tab only appears when your query uses a stats function. Confirm your query includes the stats count(*) as errors by bin(5m) line.
  • If the chart has only one bar, your runs all fell within the same 5-minute window. Wait a few minutes, run the app again, and re-run the query.

Help me understand why the Visualization tab is not showing a chart for my Logs Insights query.

You just went from "I have logs somewhere in the cloud" to answering real operational questions in seconds. These are the same queries production engineers run during outages to pinpoint root causes. Up next in the Secret Mission, you will set up a CloudWatch Alarm that emails you automatically when errors spike, so you do not have to query manually at all.

Secret mission

Set Up a CloudWatch Alarm for Error Spikes

Querying logs manually is reactive. What if CloudWatch could email you automatically when errors spike? In this secret mission, you'll wire up a proactive alerting pipeline using SNS, metric filters, and CloudWatch Alarms.

Clean Up Your Resources

Clean Up Your Resources

Decide whether to keep your resources running, pause them to come back later, or delete them entirely. This project uses AWS CloudWatch, which stays within the free tier for light usage. No ongoing charges are expected.

Resources you used:

  • CloudWatch Log Group: /learning/order-processor (free up to 5 GB/month ingestion).
  • CloudWatch Logs Metric Filter on /learning/order-processor (no additional charge).
  • CloudWatch Alarm for error spike detection (free tier covers 10 alarms).
  • SNS Topic: error-alerts with email subscription (free tier covers 1,000 email notifications/month).
  • Local project directory with order_processor.py, requirements.txt, and Python virtual environment.

Keep everything running

No action needed. Choose this if you plan to keep experimenting with your order processor or CloudWatch queries.

  • All resources stay within the AWS Free Tier for light usage.
  • CloudWatch Logs only charges for ingestion and storage. Your small log volume is well within the 5 GB/month free tier allowance.
  • The CloudWatch Alarm counts toward your 10 free alarms per month.
  • Your local Python virtual environment and project files remain on your Desktop.

Pause - I'll come back to this later

Shut down running processes to free up memory. Your cloud resources and local files stay in place for when you return.

  • Deactivate your Python virtual environment by running deactivate in your terminal.
  • No cloud-side action is needed. CloudWatch Logs only incurs charges on ingestion. Since you are not running the app, no new logs are being created.
  • When you return, reactivate your virtual environment with source venv/bin/activate (macOS/Linux) or venv\Scripts\activate (Windows).

Delete - I don't want to use this again

Remove all project resources and start fresh. Follow each step below to clean up both cloud and local resources.

Delete the CloudWatch Alarm
  • In the CloudWatch console, choose Alarms in the navigation pane.
  • Select the checkbox next to your error spike alarm.
  • Choose Actions, then Delete.
  • Choose Delete to confirm.
Delete the CloudWatch Log Group
  • In the CloudWatch console, choose Log groups in the navigation pane.
  • Select the checkbox next to /learning/order-processor.
  • Choose Actions, then Delete.
  • Confirm the deletion. This also removes the metric filter attached to the log group.
Delete the SNS topic
  • Open the Amazon SNS console.
  • In the left navigation pane, choose Topics.
  • Select the error-alerts topic, then choose Delete.
  • In the Delete topic dialog box, enter delete me and choose Delete.
Remove the IAM policy
  • Open the IAM console and choose Users in the navigation pane.
  • Select your IAM user.
  • Under the Permissions tab, find CloudWatchLogsFullAccess.
  • Choose Remove to detach the policy from your user.
Delete local project files
  • Deactivate your virtual environment by running:
deactivate
  • Delete the project folder from your Desktop by running:
cd ~/Desktop
rm -rf order-processor

On Windows, use this command instead:

cd ~\Desktop
rmdir /s /q order-processor
  • Confirm the folder is gone by running ls (macOS/Linux) or dir (Windows) and verifying the project folder no longer appears.

Nice Work!

Nice Work!

Nice work! You just built a production-grade error tracking pipeline that catches, classifies, and ships Python errors to AWS CloudWatch for persistent querying and analysis.

You've learned how to:

  • Built a Python application with a custom exception hierarchy (PaymentError, InventoryError, ShippingError) and structured try/except handling at multiple levels.
  • Replaced naive print() debugging with Python's logging module using JSON formatting, log levels, and contextual fields.
  • Connected your local application to CloudWatch using the watchtower library, establishing a persistent centralized log pipeline.
  • Queried and visualized error patterns using CloudWatch Logs Insights, answering "what type of errors happen most?" and "when did errors spike?".
  • Secret Mission: Created a proactive alerting pipeline with CloudWatch Metric Filters, Alarms, and SNS email notifications.

Ready to quiz yourself?