Guide Python Intermediate

Python File Processing Lab

Filter server records into a CSV report and handle empty, invalid, or missing input.

2 min read

Prerequisites

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

Sample input

Save as servers.csv:

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

Task

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

Run python filter_servers.py.

Expected output

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

name,server,status
web-03,prod,critical
HintUse DictReader and DictWriter with explicit encoding and newline handling. Validate fieldnames before creating the output file.
Show solution
import csv
import sys
from pathlib import Path


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


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


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

Check your work

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

InputExpected behavior
Sample aboveOne critical record, exit 0.
Header onlyHeader-only output, count 0.
Missing fileError on stderr, exit 1, no report.
Wrong headers or incomplete rowError, exit 1, no partial report.
Existing outputError; existing output unchanged.

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

Knowledge check and challenge

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

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

Add More Questions to This Guide

Know a question that should be here? Share it and help the community!

Open Google Form