Build a Resilient Ticket Pipeline

Introduction

30 Second Summary

A support team can have a full queue of useful requests when one customer number contains a typo. If that typo stops the entire queue, every clean request waits too.

In this project, you will build a Python data pipeline that enriches public API records with service tiers from a CSV file. The finished pipeline protects its JSON output by isolating malformed records.

What You'll Build

You will run one command to watch 100 public posts become a validated output/clean_tickets.json file.

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

• A live enrichment run that converts 100 public posts into service-tiered support tickets.
• Resilient streaming that exposes malformed records through warning logs without blocking valid Ticket objects.
• A tested JSON export at output/clean_tickets.json that is backed by four passing automated tests.
• Secret Mission: Add a local JSON cache that keeps the pipeline useful when the API is unavailable.

Are there any prerequisites?

Basic Python syntax is enough because the guide introduces functions, classes, and type hints as they appear. You will use Python 3.10+ with Visual Studio Code on Linux.

Before We Start

This is your moment to lock in the pipeline you are about to build. You will also define why resilient processing matters before the hands-on work begins.

Set Up the Pipeline Workspace

Your support-ticket pipeline needs a predictable place to grow. An isolated virtual environment keeps pytest separate from packages used by your other Python projects.

Here, you will create the workspace structure in Visual Studio Code. You will activate the environment before installing the pinned test dependency.

In this step, get ready to:

• Create the folders and files for the pipeline workspace.
• Create an isolated Linux virtual environment.
• Install pytest 9.1.1 and confirm its version.

Create the workspace structure

A workspace keeps the pipeline code beside its input data. The Explorer shows every project artifact in one place.

• Use your Linux application launcher to start Visual Studio Code.

You will see the Visual Studio Code window.

• Select File from the menu bar.
• Select Open Folder....

The folder picker opens so you can choose where the workspace lives.

• Select your Desktop location.
• Select New Folder.

The folder name field is ready for your project name.

• Type resilient-ticket-pipeline into the folder name field and press Enter.

The new resilient-ticket-pipeline folder appears in the folder picker.

• Select the resilient-ticket-pipeline folder.

Visual Studio Code may show a Workspace Trust dialog when you open the folder. You created this empty folder yourself, so you can trust its contents.

• Select Select Folder.
• Confirm that you trust the folder if the Workspace Trust dialog appears.

Good start. You will see resilient-ticket-pipeline at the top of the Explorer.

• Select the New Folder control in the Explorer.
• Type data and press Enter.

You will see the empty data folder in the Explorer.

• Select the New Folder control again.
• Type output and press Enter.

You will see the empty output folder beside data.

• Select the New Folder control again.
• Type tests and press Enter.

You will see the empty tests folder in the workspace.

• Select the New File... control in the Explorer.
• Type pipeline.py and press Enter.

The empty pipeline.py file opens in the editor.

• Select the New File... control again.
• Type requirements.txt and press Enter.

You will see both empty files beneath the three folders in the Explorer. The starter workspace structure is now in place.

Create and activate the virtual environment

The virtual environment stores this project's Python packages inside .venv. Activating it makes the workspace use those isolated packages.

• Select View from the menu bar.
• Select Terminal.

The integrated terminal opens at the resilient-ticket-pipeline workspace folder.

• Create the project virtual environment by running this command:

python3 -m venv .venv

What does this command do?

The venv module creates an isolated Python installation inside .venv. Project dependencies installed there stay separate from your system Python.

• Confirm that the .venv directory appears in the Explorer.

The new directory confirms that Python created the environment on disk.

• Activate the virtual environment in the workspace terminal by running this command:

source .venv/bin/activate

What does activation change?

Activation updates the current terminal session to use the Python executable inside .venv. Commands in this terminal now use the project's isolated packages.

Before you check, where do you expect the active Python path to end?

• Check the active Python executable by running this command:

which python

What does this check prove?

The which command reports the executable your terminal uses. A path inside .venv proves that activation worked.

You will see a path ending in .venv/bin/python. That path proves your pipeline now has an isolated Python environment.

Seeing a different Python path?

• Confirm that the terminal prompt points to the resilient-ticket-pipeline folder.
• Repeat the activation command from this substep if the path does not end in .venv/bin/python.

Get help with activating the virtual environment in my workspace terminal.

Pin and verify pytest

A dependency pin gives every later test run the same pytest release. The requirements.txt file records that exact version.

• Select requirements.txt in the Explorer.
• Add the pinned dependency by pasting this line:

pytest==9.1.1

What does this dependency pin do?

The pytest==9.1.1 entry tells pip to install exactly version 9.1.1. This keeps the test environment reproducible.

• Save requirements.txt.
• Install the pinned project dependency by running this command:

python3 -m pip install -r requirements.txt

What does this installation command do?

The -r option tells pip to read the dependency list from requirements.txt. The active environment receives the pinned pytest release.

The installation output completes with pytest 9.1.1 available inside the virtual environment.

Installation did not complete?

• Confirm that the active Python path still ends in .venv/bin/python.
• Retry the installation after restoring your network connection if the download was interrupted.

Get help with installing pytest from requirements.txt inside my active virtual environment.

Before you check, which pytest version do you expect this environment to report?

• Verify the installed pytest version by running this command:

pytest --version

What does this version check prove?

The version check runs pytest from the active environment. Its output confirms whether the dependency pin was installed correctly.

✔️ I see the pinned version

You will see pytest 9.1.1 in the terminal. Your isolated test environment is ready for the pipeline.

ⓧ I see an older version

An older version means the active environment does not match the dependency pin. Reinstall the requirement before checking again.

• Reinstall the pinned dependency and repeat the version check by running these commands:

python3 -m pip install -r requirements.txt
pytest --version

What does this repair do?

The first command applies the version recorded in requirements.txt. The second command confirms the installed result.

You will see pytest 9.1.1 after the installation completes.

Still seeing an older version?

Get help with finding why my active environment still reports an older pytest version.

ⓧ The pytest command is missing

The terminal needs the virtual environment activated before it can use the installed pytest command.

• Activate the environment and install the dependency before repeating the check with these commands:

source .venv/bin/activate
python3 -m pip install -r requirements.txt
pytest --version

What does this recovery do?

The first command activates the project environment. The remaining commands install pytest from the dependency file before checking its version.

You will see pytest 9.1.1 when the environment and dependency are ready.

Still cannot run pytest?

Get help with finding why pytest is unavailable after activating .venv.

✔️ Awesome, I've got everything!

Your workspace contains the required starter artifacts. Double check that requirements.txt is saved.

ⓧ I'd like to double check the full code

Compare your saved requirements.txt file with this complete reference.

pytest==9.1.1

• Confirm that pipeline.py is empty.
• Confirm that data is an empty directory for CSV input files.
• Confirm that output is an empty directory for generated JSON exports.
• Confirm that tests is an empty directory for pytest test modules.

Your workspace is ready for repeatable Python development. Next, you will build a small CSV enrichment pipeline and watch malformed input expose its first weakness.

Build a Naive Pipeline

Your isolated workspace is ready. Now you can turn the empty files into a small pipeline that produces visible ticket records.

Raw CSV values arrive as text. This first version converts each customer ID directly so you can test that assumption against realistic input.

In this step, get ready to:

• Create a local customer-tier lookup with two valid rows.
• Run a naive enrichment pipeline that prints two sample tickets.
• Challenge the direct conversion with a malformed customer ID.

Create the local CSV input

The tier file maps each customer ID to the service level that should appear on their enriched ticket.

• Select the data folder in the Visual Studio Code file tree.
• Use the new-file control above the file tree.
• Name the file customer_tiers.csv.
• Add the header plus two valid customer-tier rows by copying this data into the file:

customer_id,service_tier
1,priority
2,standard

How Is This CSV Shaped?

• The first row names the customer_id field and the service_tier field.
• The next two rows assign priority to customer 1.
• Customer 2 receives the standard tier.

• Save data/customer_tiers.csv.
• Confirm the editor shows one header row above two customer records.

Can't Find the CSV File?

• Expand the data folder in the Visual Studio Code file tree.
• Check that the filename ends with .csv.
• Move the file into data if it appears beside pipeline.py.

Help me locate or create customer_tiers.csv inside the data folder.

Run the first enrichment

This prototype uses two local post dictionaries so you can see the transformation without depending on a network request. The lookup function reads the CSV before the final loop enriches each post.

• Select pipeline.py in the Visual Studio Code file tree.
• Build the runnable prototype by copying this code into the empty file:

import csv
from pathlib import Path

posts = [
    {"id": 1, "userId": 1, "title": "Help", "body": "Printer is offline"},
    {"id": 2, "userId": 2, "title": "", "body": "No title"},
]

def load_tiers(path: Path) -> dict[int, str]:
    # Read each CSV row into a dictionary before converting the customer ID.
    tiers: dict[int, str] = {}
    with path.open(newline="", encoding="utf-8") as source:
        reader = csv.DictReader(source)
        for row in reader:
            customer_id = int(row["customer_id"])
            tiers[customer_id] = row["service_tier"]
    return tiers

# Join each local post to its service tier and print the enriched record.
tiers = load_tiers(Path("data/customer_tiers.csv"))
for post in posts:
    print(
        {
            "ticket_id": post["id"],
            "customer_id": post["userId"],
            "service_tier": tiers.get(post["userId"], "standard"),
            "title": post["title"],
            "description": post["body"],
        }
    )

What Does This Code Do?

• The posts list acts as a small local source of incoming support tickets.
• The dict[int, str] type hint describes a dictionary with integer keys and text values.
• The csv.DictReader call uses the CSV header to create one dictionary for each row.
• The int() call assumes every customer_id value can become an integer.
• The final loop adds the matching tier before printing each enriched dictionary.

• Save pipeline.py.

Before you run the pipeline, what service tier do you expect each customer to receive?

• Return to the activated workspace terminal from earlier.
• Run the local pipeline with this command:

python pipeline.py

The terminal prints two enriched dictionaries. Customer 1 has the priority tier.

Customer 2 has the standard tier. You now have a working baby pipeline that turns two inputs into recognisable ticket records.

Don't See Two Tickets?

• Confirm the terminal path ends with resilient-ticket-pipeline.
• Check that customer_tiers.csv is inside the data folder.
• Compare the indentation in pipeline.py with the code above.

Help me diagnose why python pipeline.py does not print two enriched ticket dictionaries.

Challenge the direct conversion

Incoming data does not always match the shape your code expects. The third CSV row models a customer ID that contains text instead of digits.

• Return to data/customer_tiers.csv in the Visual Studio Code file tree.
• Place your cursor below the 2,standard row.
• Add the malformed row by copying this line into the file:

not-a-number,priority

What Changed?

The new row preserves the normal two-column shape. Its customer_id value now contains not-a-number.

The Python code remains unchanged. This isolates the effect of unexpected input.

• Save data/customer_tiers.csv.

Before you rerun the pipeline, do you think both tickets will still print?

• Rerun the unchanged pipeline with this command:

python pipeline.py

The terminal now shows a ValueError that mentions not-a-number. The pipeline stops before it can print either enriched ticket.

This Failure Is Intentional

The direct int() conversion cannot turn not-a-number into an integer. The exception interrupts load_tiers before it returns the valid rows.

Keep the malformed row in place. It gives the next version a real failure case to handle.

✔️ I see the expected ValueError

Your naive pipeline has reached the intended failure. Save both files with the malformed row still present.

ⓧ I'd like to double check the full code

Compare these complete files with your versions in Visual Studio Code.

customer_id,service_tier
1,priority
2,standard
not-a-number,priority

import csv
from pathlib import Path

posts = [
    {"id": 1, "userId": 1, "title": "Help", "body": "Printer is offline"},
    {"id": 2, "userId": 2, "title": "", "body": "No title"},
]

def load_tiers(path: Path) -> dict[int, str]:
    # Read each CSV row into a dictionary before converting the customer ID.
    tiers: dict[int, str] = {}
    with path.open(newline="", encoding="utf-8") as source:
        reader = csv.DictReader(source)
        for row in reader:
            customer_id = int(row["customer_id"])
            tiers[customer_id] = row["service_tier"]
    return tiers

# Join each local post to its service tier and print the enriched record.
tiers = load_tiers(Path("data/customer_tiers.csv"))
for post in posts:
    print(
        {
            "ticket_id": post["id"],
            "customer_id": post["userId"],
            "service_tier": tiers.get(post["userId"], "standard"),
            "title": post["title"],
            "description": post["body"],
        }
    )

You found the weak point by making the first version face messy input. Next, you will isolate that bad record so valid tickets can keep moving through the pipeline.

Validate and Stream Records Safely

The naive pipeline reached useful sample data. One malformed customer ID then stopped every later record.

This step isolates that failure with validation. Custom exceptions give rejected records a meaningful name.

A generator lets each valid Ticket continue separately. Warning logging keeps skipped input visible.

In this step, get ready to:

• Define typed tickets plus meaningful pipeline errors.
• Validate each customer-tier row before accepting it.
• Stream valid tickets while logging rejected records.

Name invalid records clearly

A frozen dataclass gives every accepted ticket the same typed structure. Custom exceptions separate record-level problems from source-level failures.

• In pipeline.py, replace the current import section plus the declarations above the naive load_tiers function with the following code:

import csv
import logging
from dataclasses import dataclass
from pathlib import Path

ALLOWED_TIERS = {"standard", "priority"}
DEFAULT_TIER = "standard"

logger = logging.getLogger(__name__)


class RecordValidationError(ValueError):
    """Raised when one input record cannot become a valid domain record."""


class PipelineError(RuntimeError):
    """Raised when an input source cannot be read safely."""


@dataclass(frozen=True)
class Ticket:
    ticket_id: int
    customer_id: int
    service_tier: str
    title: str
    description: str

What Do These Declarations Add?

• The ALLOWED_TIERS set defines the accepted service tiers.
• The DEFAULT_TIER value covers customers without a matching CSV entry.
• The RecordValidationError class identifies one rejected input record.
• The PipelineError class is reserved for failures that affect an entire input source.
• The frozen Ticket class gives accepted records a consistent shape.

• Save pipeline.py.

Before you rerun the script, consider whether these new record types have changed the old direct conversion.

• Confirm that the original conversion still controls the malformed row by running:

python pipeline.py

What Does This Check Prove?

The new declarations load successfully. The existing direct int conversion still raises a ValueError for not-a-number.

Does the Script Fail Earlier?

Compare the import names plus declaration indentation against the code above. A missing parenthesis or inconsistent indentation can stop Python before it reaches the CSV row.

Ask for help with the declaration error.

Skip malformed tier rows

Row parsing needs a clear boundary around one CSV record. The parser either returns a validated pair or raises RecordValidationError with the row number.

• Add the following parse_tier_row function above load_tiers in pipeline.py.

def parse_tier_row(row: dict[str, str | None], line_number: int) -> tuple[int, str]:
    try:
        raw_customer_id = row["customer_id"]
        raw_tier = row["service_tier"]
    except KeyError as error:
        raise RecordValidationError(
            f"row {line_number}: missing column {error.args[0]}"
        ) from error

    try:
        customer_id = int(raw_customer_id)
    except (TypeError, ValueError) as error:
        raise RecordValidationError(
            f"row {line_number}: customer_id must be a positive integer"
        ) from error

    if customer_id <= 0:
        raise RecordValidationError(
            f"row {line_number}: customer_id must be a positive integer"
        )
    if not isinstance(raw_tier, str):
        raise RecordValidationError(f"row {line_number}: service_tier is missing")

    tier = raw_tier.strip().lower()
    if tier not in ALLOWED_TIERS:
        raise RecordValidationError(
            f"row {line_number}: service_tier must be standard or priority"
        )

    return customer_id, tier

How Does Row Validation Work?

• The first try block checks that both required CSV columns exist.
• The second try block translates failed customer ID conversion into a domain-specific error.
• The positive-number check rejects zero plus negative customer IDs.
• The tier check accepts only values from ALLOWED_TIERS.
• The returned tuple contains data that the rest of the pipeline can trust.

• Save pipeline.py.
• Confirm that Visual Studio Code shows no syntax error on the new function.

Seeing a Syntax Error in the Parser?

Check that each except aligns with its matching try. Check that the final return remains inside the function.

Ask for help with the parser structure.

The loader can now treat one invalid row as a local problem. Catching the custom exception inside the loop lets later rows continue.

• Replace the naive load_tiers function in pipeline.py with this resilient version:

def load_tiers(path: Path) -> dict[int, str]:
    tiers: dict[int, str] = {}

    with path.open(newline="", encoding="utf-8") as source:
        reader = csv.DictReader(source)
        for line_number, row in enumerate(reader, start=2):
            try:
                customer_id, tier = parse_tier_row(row, line_number)
            except RecordValidationError as error:
                logger.warning("Skipping tier row: %s", error)
                continue
            tiers[customer_id] = tier

    return tiers

Why Does the Pipeline Continue?

• The enumerate call starts at line 2 because line 1 contains the CSV header.
• The except RecordValidationError block catches one rejected row.
• The warning records which row was skipped.
• The continue statement advances to the next CSV row.
• Only validated customer IDs enter the tiers dictionary.

• In data/customer_tiers.csv, add 3,priority as the next row after not-a-number,priority.
• Save data/customer_tiers.csv.
• Save pipeline.py.

Before you run the loader, predict whether customer 3 can pass through after the malformed row.

• Test the row-level recovery by running:

python pipeline.py

What Should the Loader Show?

The terminal prints a warning beginning Skipping tier row: row 4. The script continues past the malformed row without a traceback.

The warning identifies rejected input without hiding it. Customer 3 still reaches the tier dictionary.

That is the first resilience win. The malformed row remains in the CSV as a repeatable test case.

Does the Malformed Row Still Stop the Script?

Make sure load_tiers calls parse_tier_row inside its try block. The old direct conversion must no longer remain inside the loader.

Ask for help with the row recovery logic.

Stream valid tickets one at a time

CSV rows are only one input boundary. Each post also needs validation before it becomes a Ticket.

• Add the following build_ticket function below load_tiers in pipeline.py.

def build_ticket(post: dict[str, object], tiers: dict[int, str]) -> Ticket:
    try:
        raw_ticket_id = post["id"]
        raw_customer_id = post["userId"]
        raw_title = post["title"]
        raw_body = post["body"]
    except KeyError as error:
        raise RecordValidationError(
            f"post is missing field {error.args[0]}"
        ) from error

    try:
        ticket_id = int(raw_ticket_id)
        customer_id = int(raw_customer_id)
    except (TypeError, ValueError) as error:
        raise RecordValidationError("post id and userId must be integers") from error

    if not isinstance(raw_title, str) or not raw_title.strip():
        raise RecordValidationError("post title must be a non-empty string")

    if not isinstance(raw_body, str) or not raw_body.strip():
        raise RecordValidationError("post body must be a non-empty string")

    return Ticket(
        ticket_id=ticket_id,
        customer_id=customer_id,
        service_tier=tiers.get(customer_id, DEFAULT_TIER),
        title=raw_title.strip(),
        description=raw_body.strip(),
    )

How Does a Post Become a Ticket?

• The first validation block requires the four source fields used by a ticket.
• The ID conversion produces integer identifiers.
• The text checks reject blank titles plus blank descriptions.
• The tier lookup uses DEFAULT_TIER when the CSV has no matching customer.
• The return statement creates one immutable Ticket.

• Save pipeline.py.
• Confirm that Visual Studio Code shows no syntax error on build_ticket.

Seeing an Error in build_ticket?

Check that every field passed to Ticket matches the dataclass field names. Check that the closing parenthesis aligns with return Ticket(.

Ask for help with ticket validation.

A generator produces one validated ticket whenever the caller asks for the next item. The pipeline avoids building another full list of ticket objects.

• Add the following iter_tickets function below build_ticket in pipeline.py.

def iter_tickets(posts: list[object], tiers: dict[int, str]):
    for position, post in enumerate(posts, start=1):
        if not isinstance(post, dict):
            logger.warning("Skipping post %s: expected an object", position)
            continue

        try:
            yield build_ticket(post, tiers)
        except RecordValidationError as error:
            logger.warning("Skipping post %s: %s", position, error)

Why Use yield Here?

• The loop checks each post independently.
• A non-dictionary post produces a warning before the loop continues.
• The yield statement passes one valid ticket to the caller.
• A rejected post produces a warning without stopping later posts.

• Replace the naive sample enrichment loop at the bottom of pipeline.py with a loop that prints each ticket from iter_tickets(posts, tiers).
• Save pipeline.py.

Before you run the finished local pipeline, predict which records will reach the terminal.

• Verify the resilient pipeline by running:

python pipeline.py

What Should the Pipeline Show?

• A warning begins with Skipping tier row: row 4.
• Each accepted sample appears as a Ticket object.
• No traceback follows the warning.

You have turned one fatal conversion into a visible rejection. Valid records now keep moving through the pipeline.

No Ticket Objects in the Terminal?

Make sure the sample list from the previous step is still named posts. Make sure the tier dictionary is still named tiers.

Check that the sample loop consumes iter_tickets(posts, tiers). The loop must print each yielded ticket.

Ask for help with the generator output.

✔️ Awesome, I've got everything!

Your resilient definitions are saved in pipeline.py. Keep the malformed CSV row in place for future checks.

ⓧ I'd like to double check the full code

Compare the reusable definitions in your pipeline.py file against this reference. Keep the small local sample runner from the previous step below these definitions for the terminal check.

import csv
import logging
from dataclasses import dataclass
from pathlib import Path

ALLOWED_TIERS = {"standard", "priority"}
DEFAULT_TIER = "standard"

logger = logging.getLogger(__name__)


class RecordValidationError(ValueError):
    """Raised when one input record cannot become a valid domain record."""


class PipelineError(RuntimeError):
    """Raised when an input source cannot be read safely."""


@dataclass(frozen=True)
class Ticket:
    ticket_id: int
    customer_id: int
    service_tier: str
    title: str
    description: str


def parse_tier_row(row: dict[str, str | None], line_number: int) -> tuple[int, str]:
    try:
        raw_customer_id = row["customer_id"]
        raw_tier = row["service_tier"]
    except KeyError as error:
        raise RecordValidationError(
            f"row {line_number}: missing column {error.args[0]}"
        ) from error

    try:
        customer_id = int(raw_customer_id)
    except (TypeError, ValueError) as error:
        raise RecordValidationError(
            f"row {line_number}: customer_id must be a positive integer"
        ) from error

    if customer_id <= 0:
        raise RecordValidationError(
            f"row {line_number}: customer_id must be a positive integer"
        )

    if not isinstance(raw_tier, str):
        raise RecordValidationError(f"row {line_number}: service_tier is missing")

    tier = raw_tier.strip().lower()
    if tier not in ALLOWED_TIERS:
        raise RecordValidationError(
            f"row {line_number}: service_tier must be standard or priority"
        )

    return customer_id, tier


def load_tiers(path: Path) -> dict[int, str]:
    tiers: dict[int, str] = {}

    with path.open(newline="", encoding="utf-8") as source:
        reader = csv.DictReader(source)
        for line_number, row in enumerate(reader, start=2):
            try:
                customer_id, tier = parse_tier_row(row, line_number)
            except RecordValidationError as error:
                logger.warning("Skipping tier row: %s", error)
                continue
            tiers[customer_id] = tier

    return tiers


def build_ticket(post: dict[str, object], tiers: dict[int, str]) -> Ticket:
    try:
        raw_ticket_id = post["id"]
        raw_customer_id = post["userId"]
        raw_title = post["title"]
        raw_body = post["body"]
    except KeyError as error:
        raise RecordValidationError(
            f"post is missing field {error.args[0]}"
        ) from error

    try:
        ticket_id = int(raw_ticket_id)
        customer_id = int(raw_customer_id)
    except (TypeError, ValueError) as error:
        raise RecordValidationError("post id and userId must be integers") from error

    if not isinstance(raw_title, str) or not raw_title.strip():
        raise RecordValidationError("post title must be a non-empty string")

    if not isinstance(raw_body, str) or not raw_body.strip():
        raise RecordValidationError("post body must be a non-empty string")

    return Ticket(
        ticket_id=ticket_id,
        customer_id=customer_id,
        service_tier=tiers.get(customer_id, DEFAULT_TIER),
        title=raw_title.strip(),
        description=raw_body.strip(),
    )


def iter_tickets(posts: list[object], tiers: dict[int, str]):
    for position, post in enumerate(posts, start=1):
        if not isinstance(post, dict):
            logger.warning("Skipping post %s: expected an object", position)
            continue

        try:
            yield build_ticket(post, tiers)
        except RecordValidationError as error:
            logger.warning("Skipping post %s: %s", position, error)

How to Use This Reference

This file contains the reusable validation plus streaming code introduced in this step. The next step connects these definitions to API ingestion plus JSON export.

Your local records now survive malformed input. Next, you will replace the sample posts with API data before streaming clean tickets into JSON.

Fetch API Data and Export JSON

Your validation layer now skips malformed records while valid tickets continue through the pipeline.

Hard-coded posts only prove the transformation with fixed data. This step connects the pipeline to the JSONPlaceholder API before streaming the enriched tickets into a valid JSON file.

In this step, get ready to:

• Fetch 100 posts from JSONPlaceholder.
• Stream enriched tickets into a JSON array.
• Run the full pipeline and validate the exported file.

Fetch posts from the API

A network request creates a new failure boundary. The pipeline needs to translate connection problems or malformed responses into the existing PipelineError exception.

• In pipeline.py, replace the import block at the top with this expanded block:

import csv
import json
import logging
import urllib.error
import urllib.request
from dataclasses import dataclass
from pathlib import Path

API_URL = "https://jsonplaceholder.typicode.com/posts"

What do these additions provide?

• The json module decodes the API response.
• The urllib.request module opens the HTTP connection.
• The urllib.error module provides the network exception handled by the pipeline.
• The API_URL constant keeps the posts endpoint in one reusable location.

• Save pipeline.py.

You should see API_URL above ALLOWED_TIERS at the top of the file.

Seeing import errors?

Compare the import names against the block above. These modules come with Python, so this change does not require another installation.

Check that the API_URL value uses matching double quotes.

Help me fix the new imports or API_URL in pipeline.py.

• In pipeline.py, add the API fetcher below iter_tickets by copying this code:

def fetch_posts(url: str = API_URL) -> list[object]:
    logger.info("Fetching posts from %s", url)

    try:
        with urllib.request.urlopen(url, timeout=10) as response:
            payload = json.load(response)
    except (urllib.error.URLError, json.JSONDecodeError) as error:
        raise PipelineError(f"Could not fetch valid JSON from {url}") from error

    if not isinstance(payload, list):
        raise PipelineError("API response must be a JSON array")

    return payload

How does fetching stay resilient?

• The request uses a 10-second timeout so the pipeline does not wait indefinitely for a response.
• The try block catches network failures or invalid JSON.
• The response check requires a list because the posts endpoint returns a JSON array.
• The function returns the decoded posts for the existing generator to process.

Why urllib.request?

The standard-library client keeps the project focused on pipeline mechanics without adding another runtime dependency. A higher-level library such as Requests would add package setup that this small GET request does not need.

• Save pipeline.py.

You should now see fetch_posts between iter_tickets and the output functions you add next.

Seeing an error in fetch_posts?

Check that fetch_posts starts at the left edge of the file. Accidental indentation can place it inside iter_tickets.

Confirm the except block aligns with try.

Help me correct the fetch_posts function in pipeline.py.

Write tickets as JSON

Each Ticket is a dataclass instance. The JSON encoder needs a dictionary containing values it can serialize.

• In pipeline.py, add this conversion function below fetch_posts by copying the code:

def ticket_to_dict(ticket: Ticket) -> dict[str, object]:
    return {
        "ticket_id": ticket.ticket_id,
        "customer_id": ticket.customer_id,
        "service_tier": ticket.service_tier,
        "title": ticket.title,
        "description": ticket.description,
    }

What does this conversion preserve?

The dictionary keeps the five fields from Ticket. Each field becomes a named property in the exported JSON object.

• Save pipeline.py.

You should see all five Ticket fields represented once in the returned dictionary.

Missing a ticket field?

Compare the dictionary keys with the fields in the Ticket dataclass. The names must match the target JSON structure exactly.

Help me compare ticket_to_dict with the Ticket dataclass.

The writer consumes the generator one ticket at a time. It inserts commas between objects so the finished file remains one valid JSON array.

• In pipeline.py, add the writer below ticket_to_dict by copying this code:

def write_tickets(tickets, output_path: Path) -> int:
    output_path.parent.mkdir(parents=True, exist_ok=True)
    count = 0

    with output_path.open("w", encoding="utf-8") as target:
        target.write("[\n")
        for ticket in tickets:
            if count:
                target.write(",\n")
            json.dump(ticket_to_dict(ticket), target, ensure_ascii=False)
            count += 1
        target.write("\n]\n")

    return count

How does incremental export work?

• The first line creates the destination directory when it is missing.
• The opening bracket starts the JSON array before the generator produces a ticket.
• The loop converts each ticket immediately before writing it.
• The returned count records how many tickets reached the output file.

• Save pipeline.py.
• Confirm the expanded file still runs by using this command:

python pipeline.py

What should I see?

The current sample runner should still print the valid Ticket objects from the previous step. This confirms the new output functions load without breaking the existing generator.

Did the sample pipeline stop running?

Check the indentation inside write_tickets. The closing bracket write belongs inside the with block after the loop.

Check that return count aligns with the with statement.

Help me fix write_tickets after the existing sample pipeline stopped running.

Connect and verify the full pipeline

The input and output stages are ready. The main function now connects CSV loading, API fetching, generator processing, and JSON export.

• Delete the temporary sample-data runner from the bottom of pipeline.py.
• Add the complete runtime flow at the bottom of pipeline.py by copying this code:

def main() -> None:
    logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")

    tier_path = Path("data/customer_tiers.csv")
    output_path = Path("output/clean_tickets.json")

    tiers = load_tiers(tier_path)
    posts = fetch_posts()
    count = write_tickets(iter_tickets(posts, tiers), output_path)

    logger.info("Exported %s tickets to %s", count, output_path)


if __name__ == "__main__":
    main()

How does the pipeline connect?

• Logging makes the skipped CSV row and completed export visible.
• The two paths identify the existing CSV input and the new JSON output.
• The API posts flow directly into iter_tickets before write_tickets consumes them.
• The final log reports the number of exported tickets.

• Save pipeline.py.

Before you run the pipeline, do you expect the malformed CSV row to stop the API export?

• Run the connected pipeline from the workspace terminal with this command:

python pipeline.py

What should I see?

You should see a warning beginning WARNING Skipping tier row: row 4. The pipeline continues because load_tiers isolates the malformed row.

The final log should say Exported 100 tickets to output/clean_tickets.json. That confirms the live API records reached the output stage.

Did the API export fail?

Confirm the virtual environment remains active. Check that your internet connection can reach JSONPlaceholder.

If the failure points to pipeline.py, compare main against the block above. A misspelled function name can break the stage connection.

Help me diagnose why python pipeline.py did not export 100 tickets.

Before you validate the file, what JSON structure do you expect the command to display?

• Validate and pretty-print the exported file by running this command:

python -m json.tool output/clean_tickets.json

What should I see?

You should see a JSON array containing 100 objects. Each object includes ticket_id, customer_id, service_tier, title, and description.

Is the JSON file missing or invalid?

Confirm the export command completed before running the JSON validator. The output/clean_tickets.json file is created during that run.

Compare the opening bracket, comma logic, and closing bracket in write_tickets against the completed file below.

Help me fix an invalid output/clean_tickets.json file produced by write_tickets.

That is the full data path working. Your pipeline now fetches live records while one malformed CSV row stays safely contained.

✔️ Awesome, I've got everything!

Great. Double check that you saved pipeline.py after the successful export.

ⓧ I'd like to double check the full code

Compare your complete pipeline.py file with this version.

import csv
import json
import logging
import urllib.error
import urllib.request
from dataclasses import dataclass
from pathlib import Path

API_URL = "https://jsonplaceholder.typicode.com/posts"
ALLOWED_TIERS = {"standard", "priority"}
DEFAULT_TIER = "standard"

logger = logging.getLogger(__name__)


class RecordValidationError(ValueError):
    """Raised when one input record cannot become a valid domain record."""


class PipelineError(RuntimeError):
    """Raised when an input source cannot be read safely."""


@dataclass(frozen=True)
class Ticket:
    ticket_id: int
    customer_id: int
    service_tier: str
    title: str
    description: str


def parse_tier_row(row: dict[str, str | None], line_number: int) -> tuple[int, str]:
    try:
        raw_customer_id = row["customer_id"]
        raw_tier = row["service_tier"]
    except KeyError as error:
        raise RecordValidationError(
            f"row {line_number}: missing column {error.args[0]}"
        ) from error

    try:
        customer_id = int(raw_customer_id)
    except (TypeError, ValueError) as error:
        raise RecordValidationError(
            f"row {line_number}: customer_id must be a positive integer"
        ) from error

    if customer_id <= 0:
        raise RecordValidationError(
            f"row {line_number}: customer_id must be a positive integer"
        )

    if not isinstance(raw_tier, str):
        raise RecordValidationError(f"row {line_number}: service_tier is missing")

    tier = raw_tier.strip().lower()
    if tier not in ALLOWED_TIERS:
        raise RecordValidationError(
            f"row {line_number}: service_tier must be standard or priority"
        )

    return customer_id, tier


def load_tiers(path: Path) -> dict[int, str]:
    tiers: dict[int, str] = {}

    with path.open(newline="", encoding="utf-8") as source:
        reader = csv.DictReader(source)
        for line_number, row in enumerate(reader, start=2):
            try:
                customer_id, tier = parse_tier_row(row, line_number)
            except RecordValidationError as error:
                logger.warning("Skipping tier row: %s", error)
                continue
            tiers[customer_id] = tier

    return tiers


def build_ticket(post: dict[str, object], tiers: dict[int, str]) -> Ticket:
    try:
        raw_ticket_id = post["id"]
        raw_customer_id = post["userId"]
        raw_title = post["title"]
        raw_body = post["body"]
    except KeyError as error:
        raise RecordValidationError(
            f"post is missing field {error.args[0]}"
        ) from error

    try:
        ticket_id = int(raw_ticket_id)
        customer_id = int(raw_customer_id)
    except (TypeError, ValueError) as error:
        raise RecordValidationError("post id and userId must be integers") from error

    if not isinstance(raw_title, str) or not raw_title.strip():
        raise RecordValidationError("post title must be a non-empty string")

    if not isinstance(raw_body, str) or not raw_body.strip():
        raise RecordValidationError("post body must be a non-empty string")

    return Ticket(
        ticket_id=ticket_id,
        customer_id=customer_id,
        service_tier=tiers.get(customer_id, DEFAULT_TIER),
        title=raw_title.strip(),
        description=raw_body.strip(),
    )


def iter_tickets(posts: list[object], tiers: dict[int, str]):
    for position, post in enumerate(posts, start=1):
        if not isinstance(post, dict):
            logger.warning("Skipping post %s: expected an object", position)
            continue

        try:
            yield build_ticket(post, tiers)
        except RecordValidationError as error:
            logger.warning("Skipping post %s: %s", position, error)


def fetch_posts(url: str = API_URL) -> list[object]:
    logger.info("Fetching posts from %s", url)

    try:
        with urllib.request.urlopen(url, timeout=10) as response:
            payload = json.load(response)
    except (urllib.error.URLError, json.JSONDecodeError) as error:
        raise PipelineError(f"Could not fetch valid JSON from {url}") from error

    if not isinstance(payload, list):
        raise PipelineError("API response must be a JSON array")

    return payload


def ticket_to_dict(ticket: Ticket) -> dict[str, object]:
    return {
        "ticket_id": ticket.ticket_id,
        "customer_id": ticket.customer_id,
        "service_tier": ticket.service_tier,
        "title": ticket.title,
        "description": ticket.description,
    }


def write_tickets(tickets, output_path: Path) -> int:
    output_path.parent.mkdir(parents=True, exist_ok=True)
    count = 0

    with output_path.open("w", encoding="utf-8") as target:
        target.write("[\n")
        for ticket in tickets:
            if count:
                target.write(",\n")
            json.dump(ticket_to_dict(ticket), target, ensure_ascii=False)
            count += 1
        target.write("\n]\n")

    return count


def main() -> None:
    logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")

    tier_path = Path("data/customer_tiers.csv")
    output_path = Path("output/clean_tickets.json")

    tiers = load_tiers(tier_path)
    posts = fetch_posts()
    count = write_tickets(iter_tickets(posts, tiers), output_path)

    logger.info("Exported %s tickets to %s", count, output_path)


if __name__ == "__main__":
    main()

How to compare the full file

Check the order of the imports, constants, classes, and functions. Then confirm the main guard remains at the bottom of the file.

Your live CSV-to-JSON pipeline now produces a validated artifact from 100 API records. Next, you will use pytest to prove its success paths and failure handling remain reliable.

Test the Failure Paths

Your live CSV-to-JSON pipeline now validates records, skips malformed input, and exports clean tickets. A future edit could silently remove any of those safeguards.

Automated tests make that resilience repeatable. In this step, pytest recreates each important path with small local inputs, so the suite never depends on the live API.

In this step, get ready to:

• Prove missing fields raise the expected validation exception.
• Confirm malformed CSV rows produce warnings without stopping valid work.
• Verify generator behavior and JSON output with isolated test data.

Test validation and generator behavior

The first tests target two core promises. Invalid posts raise a meaningful exception, while the generator continues to yield valid tickets.

• In the Explorer sidebar, right-click the tests folder.
• Select New File.
• Enter test_pipeline.py as the file name.
• Populate tests/test_pipeline.py with the imports plus the missing-title test by pasting this code:

import json
import logging

import pytest

from pipeline import (
    RecordValidationError,
    Ticket,
    build_ticket,
    iter_tickets,
    load_tiers,
    write_tickets,
)


def test_build_ticket_rejects_missing_title():
    with pytest.raises(RecordValidationError, match="title"):
        build_ticket(
            {"id": 1, "userId": 1, "body": "Printer is offline"},
            {1: "priority"},
        )

What does this test prove?

• The imports expose the production functions that each test exercises directly.
• The pytest.raises() context confirms that build_ticket rejects a post with no title.
• The match check confirms that the exception message identifies the missing title.

• Save tests/test_pipeline.py.
• Run the first test from the workspace terminal by running:

pytest

What does this command check?

Pytest discovers tests in test_*.py files below the current directory. It runs the missing-title test against the local build_ticket function.

You should see the missing-title test pass. The first failure path now has a repeatable check.

Does the first test fail to run?

• Confirm that test_pipeline.py is inside the existing tests folder.
• Compare every imported name against the definitions in pipeline.py.

Ask for help with a pytest collection or import problem.

A generator processes one item at a time. This test feeds it one valid post plus one invalid post, then checks that only the valid ticket reaches the result.

• Add the generator test below test_build_ticket_rejects_missing_title in tests/test_pipeline.py by pasting:

def test_iter_tickets_keeps_valid_posts():
    posts = [
        {"id": 1, "userId": 1, "title": "Help", "body": "Printer is offline"},
        {"id": 2, "userId": 2, "title": "", "body": "No title"},
    ]

    tickets = list(iter_tickets(posts, {1: "priority"}))

    assert tickets == [
        Ticket(
            ticket_id=1,
            customer_id=1,
            service_tier="priority",
            title="Help",
            description="Printer is offline",
        )
    ]

How does this test isolate the generator?

• The posts list supplies two small dictionaries without fetching remote data.
• The call to list(iter_tickets(...)) consumes the generator so its yielded tickets can be compared.
• The assertion proves that the valid post becomes a Ticket while the empty-title post is skipped.

• Save tests/test_pipeline.py.
• Run both tests from the workspace terminal by running:

pytest

What does this run check?

Pytest reruns the exception test alongside the new generator test. Both tests use in-memory dictionaries instead of the network.

You should see both tests pass. The generator now has proof that one invalid post does not block valid output.

Does the generator test fail?

• Check that the first post includes id, userId, title, and body.
• Confirm that the second post keeps its title value empty.
• Compare the expected Ticket fields against the code block above.

Ask for help with a failing generator assertion.

Test CSV warnings in isolation

A temporary path keeps test data away from your real data/customer_tiers.csv file. Captured logs prove that load_tiers reports the malformed row before continuing.

• Add the temporary CSV test below test_iter_tickets_keeps_valid_posts in tests/test_pipeline.py by pasting:

def test_load_tiers_skips_bad_rows(tmp_path, caplog):
    csv_path = tmp_path / "tiers.csv"
    csv_path.write_text(
        "customer_id,service_tier\n1,priority\nbad,standard\n",
        encoding="utf-8",
    )
    caplog.set_level(logging.WARNING)

    tiers = load_tiers(csv_path)

    assert tiers == {1: "priority"}
    assert "Skipping tier row: row 3" in caplog.text

How does this test capture a warning?

• The tmp_path fixture provides a unique temporary directory for the test CSV file.
• The caplog fixture captures warning records produced during the test.
• The first assertion proves that the valid row survives.
• The second assertion proves that the malformed row produces the expected warning.

• Save tests/test_pipeline.py.
• Run the three tests from the workspace terminal by running:

pytest

What does this run check?

Pytest creates the temporary path before running the CSV test. It also captures the warning without changing your production data file.

You should see three passing tests. Your malformed-row behavior is now verified through both returned data and captured logging.

Does the CSV test fail?

• Confirm that the temporary CSV string includes a newline after each row.
• Check that caplog.set_level(logging.WARNING) appears before load_tiers(csv_path).
• Compare the asserted row number against the temporary CSV content.

Ask for help with the temporary CSV or captured warning.

Test JSON export and run the suite

The final test sends one in-memory Ticket through write_tickets. It reads the temporary output with json.loads to confirm that the exported file is valid JSON with the expected fields.

• Add the export test below test_load_tiers_skips_bad_rows in tests/test_pipeline.py by pasting:

def test_write_tickets_creates_valid_json(tmp_path):
    output_path = tmp_path / "output" / "tickets.json"
    ticket = Ticket(
        ticket_id=1,
        customer_id=1,
        service_tier="priority",
        title="Help",
        description="Printer is offline",
    )

    count = write_tickets(iter([ticket]), output_path)
    payload = json.loads(output_path.read_text(encoding="utf-8"))

    assert count == 1
    assert payload[0]["service_tier"] == "priority"
    assert payload[0]["title"] == "Help"

What does this export test prove?

• The temporary output path prevents the test from overwriting output/clean_tickets.json.
• The iter([ticket]) call supplies an in-memory iterator that behaves like the production ticket stream.
• The parsed payload proves that the file contains valid JSON.
• The assertions verify the exported count plus two representative fields.

• Save tests/test_pipeline.py.

✔️ Awesome, I've got everything!

Your test file now contains four isolated tests for validation, generator behavior, warning logs, and JSON export.

ⓧ I'd like to double check the full code

• Compare your complete tests/test_pipeline.py file against this reference:

import json
import logging

import pytest

from pipeline import (
    RecordValidationError,
    Ticket,
    build_ticket,
    iter_tickets,
    load_tiers,
    write_tickets,
)


def test_build_ticket_rejects_missing_title():
    with pytest.raises(RecordValidationError, match="title"):
        build_ticket(
            {"id": 1, "userId": 1, "body": "Printer is offline"},
            {1: "priority"},
        )


def test_iter_tickets_keeps_valid_posts():
    posts = [
        {"id": 1, "userId": 1, "title": "Help", "body": "Printer is offline"},
        {"id": 2, "userId": 2, "title": "", "body": "No title"},
    ]

    tickets = list(iter_tickets(posts, {1: "priority"}))

    assert tickets == [
        Ticket(
            ticket_id=1,
            customer_id=1,
            service_tier="priority",
            title="Help",
            description="Printer is offline",
        )
    ]


def test_load_tiers_skips_bad_rows(tmp_path, caplog):
    csv_path = tmp_path / "tiers.csv"
    csv_path.write_text(
        "customer_id,service_tier\n1,priority\nbad,standard\n",
        encoding="utf-8",
    )
    caplog.set_level(logging.WARNING)

    tiers = load_tiers(csv_path)

    assert tiers == {1: "priority"}
    assert "Skipping tier row: row 3" in caplog.text


def test_write_tickets_creates_valid_json(tmp_path):
    output_path = tmp_path / "output" / "tickets.json"
    ticket = Ticket(
        ticket_id=1,
        customer_id=1,
        service_tier="priority",
        title="Help",
        description="Printer is offline",
    )

    count = write_tickets(iter([ticket]), output_path)
    payload = json.loads(output_path.read_text(encoding="utf-8"))

    assert count == 1
    assert payload[0]["service_tier"] == "priority"
    assert payload[0]["title"] == "Help"

Before you run the full suite, do you expect all four tests to pass without contacting JSONPlaceholder?

• Run the full suite from the workspace terminal by running:

pytest

What does the full suite cover?

• One test checks the expected validation exception.
• One test checks that the generator keeps valid posts.
• One test checks temporary CSV processing plus warning logs.
• One test checks the exported count plus valid JSON fields.

The summary reports 4 passed. That result proves your resilience checks work without a live API call.

Does the suite report a failure?

• Read the first failed assertion to identify which test needs attention.
• Compare that test against the complete reference above.
• Confirm that you saved tests/test_pipeline.py before rerunning the suite.

Ask for help with a failing four-test suite.

Secret mission

Add an Offline JSON Cache

Your pipeline currently depends on a successful network request. In this mission, you will save successful API responses to a local JSON cache and use that cache when the API is unavailable.

Clean Up Your Resources

Clean Up Your Resources

Everything for this project lives on your computer. Keeping it creates no cloud charges.

Choose the cleanup option that fits what you want to do next.

Resources you used:

• The resilient-ticket-pipeline workspace folder on your computer.

Keep everything running

No action needed. Choose this if you want to keep developing or demonstrating the pipeline.

• Leave the resilient-ticket-pipeline workspace folder in place.
• Keep data/posts_cache.json for offline fallback runs.
• Keep output/clean_tickets.json as the latest exported pipeline artifact.
• Keep .venv inside the workspace for isolated test runs.

Pause - I'll come back to this later

There is no server process or cloud resource to stop. Your project files remain available on disk.

• Close Visual Studio Code to end the active terminal session.
• Leave the resilient-ticket-pipeline workspace folder on your computer for your next session.

Delete - I don't want to use this again

This option removes all project resources for a fresh start. The command permanently targets only the resilient-ticket-pipeline folder.

• Delete the resilient-ticket-pipeline workspace from its parent folder by running these commands:

cd ..
rm -rf resilient-ticket-pipeline

What Do These Commands Remove?

• The first command moves the terminal one folder above the workspace.
• The second command removes the named workspace folder plus every file inside it.

• Close the workspace terminal to end the activated .venv session.
• Switch back to the workspace file list in Visual Studio Code.
• Confirm that pipeline.py no longer appears in the workspace file list.
• Confirm that the data folder no longer appears in the workspace file list.

Still See the Workspace Folder?

• Check that the terminal is in the folder that directly contains resilient-ticket-pipeline.
• Check that the workspace folder name matches resilient-ticket-pipeline exactly.

Ask for help with safely removing this workspace.

Nice Work!

Nice Work!

You did it! Your resilient support-ticket data pipeline now turns raw inputs into dependable JSON output.

You learned how to:

• Built a CSV enrichment stage that combines service tiers with incoming records. Kept malformed data in the input to expose the naive conversion failure before fixing it.
• Added typed Ticket records with validation rules. Used custom exceptions to isolate bad records. Streamed valid tickets through a generator. Made skipped work visible with warning logs.
• Connected the pipeline to the JSONPlaceholder public API. Streamed 100 enriched tickets into output/clean_tickets.json. Proved each failure path with five isolated pytest tests.
• Secret Mission: Added a tested offline JSON cache through fetch_posts_with_fallback. The fallback logs a warning before loading data/posts_cache.json when the API is unavailable.

Ready to quiz yourself?