Handling Python Exceptions with Structured Logging

Hello! Last lesson, your script read mock_results.json, classified its results, and wrote decision_report.json. It worked when the input was present and correctly shaped, but a missing file or malformed JSON would stop it with a traceback. Today you’ll make those expected failures understandable while leaving unexpected programming mistakes visible.

Plan for about 45 minutes: roughly 20 minutes of video, then 20–25 minutes to update and test the local script. No AWS account or API key is needed.

Handle the failures you can anticipate

An exception is Python’s way of reporting that an operation could not complete. Opening a nonexistent file raises FileNotFoundError; parsing malformed JSON raises json.JSONDecodeError. A try block identifies the operation that might fail, and a matching except block decides what your program should do about that particular failure.

Watch Corey Schafer’s Python Tutorial: Using Try/Except Blocks for Error Handling. Pay particular attention to why a handler should name the failure it expects.

The important distinction is between recovering or reporting a known condition and concealing a bug. If the input file is absent, your script can log that fact and exit cleanly. If you mistype a variable name, catching every Exception and reporting “file missing” would send you in the wrong direction. Keep try blocks focused, and catch the exception types you have a plan for.

Valid JSON can still be unusable input. For example, {"results": "not a list"} parses successfully, but your loop expects a list. After parsing, you can check the data’s shape and raise ValueError(...) with a reason you control. That gives malformed content a different diagnosis from malformed JSON syntax.

Make each log line useful to a future operator

A traceback is valuable when debugging a bug, but it is not the only way to report normal operational failures. Logs record events such as “report written” or “input JSON invalid,” with a severity level. INFO suits successful progress; ERROR suits a run that could not produce its report.

Watch Python Logging - Tutorial from Tech With Tim for the logging module’s basic tools. The video demonstrates conventional text logs; afterward, we’ll format the same kinds of events as JSON.

A sentence such as ERROR payment failed user=usr_4421 duration=5023ms is readable, but software must pick apart its text to find the user or duration. A structured log gives those pieces separate, consistently named fields. This comparison illustrates the difference:

For your local script, a useful log event will have a UTC time, level, short event name, and a few fields relevant to that event. For example, an invalid-JSON event can include the line and column of the syntax error. Avoid logging the full research query, raw provider responses, credentials, or arbitrary exception text: those may contain sensitive data. A short, consistent event name also makes logs easier to search later.

Python’s logger.exception(...), shown in the video, is useful inside an exception handler when an authorized operator needs a traceback. You do not need a traceback for every anticipated bad input. For this script, a concise error event and a nonzero exit status are more useful.

Update the research-and-decision script

In your research-decision-api folder, keep the mock_results.json from last lesson. Replace json_demo.py with the version below. The classifier and report are familiar; the additions are input checks, targeted exception handlers, and a JSON log formatter.

import json
import logging
import os
from datetime import datetime, timezone


class JsonFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        entry = {
            "time": datetime.fromtimestamp(
                record.created, timezone.utc
            ).isoformat(timespec="seconds"),
            "level": record.levelname,
            "event": record.getMessage(),
        }
        entry.update(getattr(record, "fields", {}))

        if record.exc_info:
            entry["traceback"] = self.formatException(record.exc_info)

        return json.dumps(entry, ensure_ascii=False)


logger = logging.getLogger("research_decision")
handler = logging.StreamHandler()
handler.setFormatter(JsonFormatter())
logger.addHandler(handler)
logger.setLevel(logging.INFO)
logger.propagate = False


def classify_result(score: float, active: bool) -> str:
    if active and score >= 0.8:
        return "accepted"
    elif active and score >= 0.5:
        return "review"
    return "skipped"


def build_report(data: object, run_label: str) -> dict:
    if not isinstance(data, dict):
        raise ValueError("input must be an object")

    query = data.get("query")
    results = data.get("results")
    if not isinstance(query, str):
        raise ValueError("query must be a string")
    if not isinstance(results, list):
        raise ValueError("results must be a list")

    decisions = []
    counts = {"accepted": 0, "review": 0, "skipped": 0}

    for index, result in enumerate(results):
        if not isinstance(result, dict):
            raise ValueError(f"result {index} must be an object")

        title = result.get("title")
        score = result.get("score")
        active = result.get("active")

        if not isinstance(title, str):
            raise ValueError(f"result {index}: title must be text")
        if type(score) not in (int, float) or not 0 <= score <= 1:
            raise ValueError(f"result {index}: score must be from 0 to 1")
        if not isinstance(active, bool):
            raise ValueError(f"result {index}: active must be true or false")

        status = classify_result(score, active)
        decisions.append({"title": title, "status": status})
        counts[status] += 1

    return {
        "query": query,
        "run_label": run_label,
        "decisions": decisions,
        "counts": counts,
    }


def main() -> int:
    input_path = "mock_results.json"
    output_path = "decision_report.json"
    run_label = os.environ.get("RESEARCH_RUN_LABEL", "local")

    try:
        with open(input_path, "r", encoding="utf-8") as input_file:
            data = json.load(input_file)
    except FileNotFoundError:
        logger.error("input_missing", extra={"fields": {"file": input_path}})
        return 1
    except json.JSONDecodeError as exc:
        logger.error(
            "input_json_invalid",
            extra={"fields": {"line": exc.lineno, "column": exc.colno}},
        )
        return 1
    except UnicodeError:
        logger.error("input_encoding_invalid")
        return 1
    except OSError as exc:
        logger.error(
            "input_read_failed",
            extra={"fields": {"error_type": type(exc).__name__}},
        )
        return 1

    try:
        report = build_report(data, run_label)
    except ValueError as exc:
        # This reason was constructed by build_report, not copied from input.
        logger.error("input_shape_invalid", extra={"fields": {"reason": str(exc)}})
        return 1

    try:
        with open(output_path, "w", encoding="utf-8") as output_file:
            json.dump(report, output_file, indent=2)
    except OSError as exc:
        logger.error(
            "report_write_failed",
            extra={"fields": {"error_type": type(exc).__name__}},
        )
        return 1

    logger.info(
        "report_written",
        extra={"fields": {"file": output_path, **report["counts"]}},
    )
    return 0


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

The formatter produces one JSON object per log event. The extra={"fields": ...} argument supplies event-specific fields without assembling a free-form message. The logger supplies the level, and the formatter supplies the timestamp. The record.exc_info branch would preserve a traceback inside one JSON field if you later deliberately use exception logging; this script does not attach one to expected input errors.

Notice the boundaries of the three try blocks. The first protects reading and parsing, the second checks whether parsed data has the shape the classifier needs, and the third protects writing. This lets input_read_failed and report_write_failed mean different things. An unexpected programming error remains unhandled, so Python still shows its traceback instead of labeling it as bad input.

Run it from the VS Code PowerShell terminal in the project folder:

python .\json_demo.py

If needed, use .\.venv\Scripts\python.exe .\json_demo.py instead. With last lesson’s input, expect a single INFO JSON log line with event set to report_written and counts of one accepted, one review, and two skipped. Open decision_report.json to confirm the report itself is still written. No Tavily key or other credential appears in either output.

Check that the failure tells the truth

First, simulate a missing file in PowerShell. Temporarily rename the input, run the script, then restore it:

Rename-Item .\mock_results.json mock_results.backup.json
python .\json_demo.py
Rename-Item .\mock_results.backup.json mock_results.json

The run should emit an ERROR event named input_missing, rather than a Python traceback. To test broken JSON syntax without losing your input, make a backup and temporarily replace the file:

Copy-Item .\mock_results.json .\mock_results.backup.json
Set-Content .\mock_results.json '{"results": [}'
python .\json_demo.py
Move-Item -Force .\mock_results.backup.json .\mock_results.json

This time, expect input_json_invalid with line and column fields. A file containing valid JSON with "results": "not a list" would instead produce input_shape_invalid: parsing succeeded, but validation did not.

On these handled failures, main() returns 1, which becomes the process’s nonzero exit status; success returns 0. A failed run does not remove an older decision_report.json, so do not treat a report left from a previous run as evidence that the latest run succeeded. If a write itself fails, a partial output file may also remain. The log event and exit status tell you whether this run completed.

Where this leaves your project

You can now catch specific, anticipated exceptions, distinguish invalid JSON from incorrectly shaped data, and emit searchable JSON log events without exposing raw input or secrets. Unexpected bugs still surface as bugs rather than being hidden by a catch-all handler.

Next, you’ll call a REST API from Python. The same habits will apply there: distinguish a successful response from an HTTP error, a timeout, or invalid response JSON, and report each failure accurately.