Read CSV safely and build an auditable pipeline

File operations have their own failure modes

A file can be absent, inaccessible, empty or structurally wrong. A row inside a valid file can contain an invalid value. Those are distinct states. If a whole file cannot be read, reporting zero devices would turn a collection failure into a false inventory claim.

Use with to scope an open file so it is closed when the block finishes. Give the source encoding explicitly. For CSV files opened with Python's CSV module, use newline="" so the module can handle newline conventions correctly. A source file path should be supplied deliberately; do not assume every scheduled process uses your interactive working directory.

csv.DictReader reads column headings and supplies each row as a dictionary. It does not automatically convert percentage text to integers. Validate required headings before classifying rows. A misspelled free_pct heading is a file-contract failure, not a row whose capacity happens to be unknown.

Complete project implementation

Save this as inventory_report.py. It contains no network access and no device remediation. The default operation produces a separate CSV report from the fictional input you supply.

import argparse
import csv
from collections import Counter
from pathlib import Path

REQUIRED = {"host", "free_pct"}


def classify_row(row):
    original_host = row.get("host")
    raw = row.get("free_pct")
    host = original_host.strip() if isinstance(original_host, str) else ""
    if not host:
        return {"host": host, "source_value": raw, "free_pct": None,
                "status": "INVALID", "reason": "missing host"}
    try:
        if not isinstance(raw, str):
            raise ValueError("missing or non-text value")
        free = int(raw)
    except (ValueError, TypeError):
        return {"host": host, "source_value": raw, "free_pct": None,
                "status": "INVALID", "reason": "not an integer"}
    if not 0 <= free <= 100:
        return {"host": host, "source_value": raw, "free_pct": None,
                "status": "INVALID", "reason": "outside 0..100"}
    return {"host": host, "source_value": raw, "free_pct": free,
            "status": "REVIEW" if free < 10 else "OK", "reason": "valid measurement"}


def read_report(path):
    with path.open("r", encoding="utf-8-sig", newline="") as handle:
        reader = csv.DictReader(handle)
        fields = reader.fieldnames or []
        if len(fields) != len(set(fields)):
            raise ValueError("duplicate CSV headings")
        if not REQUIRED.issubset(fields):
            raise ValueError("required headings: host, free_pct")
        output = []
        for number, row in enumerate(reader, start=2):
            if None in row:
                raise ValueError(f"extra CSV fields on line {number}")
            result = classify_row(row)
            result["line"] = number
            output.append(result)
        if not output:
            raise ValueError("input has headings but no data rows")
    return output


def write_report(path, rows):
    # Exclusive creation prevents silent replacement of a previous result.
    with path.open("x", encoding="utf-8", newline="") as handle:
        writer = csv.DictWriter(handle, fieldnames=["line", "host", "source_value",
                               "free_pct", "status", "reason"])
        writer.writeheader()
        writer.writerows(rows)


def main():
    parser = argparse.ArgumentParser(description="Read-only inventory classification")
    parser.add_argument("source", type=Path)
    parser.add_argument("report", type=Path)
    args = parser.parse_args()
    if args.source.resolve() == args.report.resolve():
        parser.error("source and report must be different files")
    try:
        rows = read_report(args.source)
        write_report(args.report, rows)
    except (OSError, ValueError) as error:
        parser.exit(2, f"Report failed: {error}\n")
    counts = Counter(row["status"] for row in rows)
    print(f"Rows={len(rows)}; INVALID={counts['INVALID']}; "
          f"REVIEW={counts['REVIEW']}; OK={counts['OK']}")


if __name__ == "__main__":
    main()

Create inventory.csv with a header host,free_pct and five rows: LAB-A,9; LAB-B,10; LAB-C,0; LAB-D,unknown; LAB-E,101. Each row goes on its own line. Run python inventory_report.py inventory.csv report.csv. The expected summary is Rows=5; INVALID=2; REVIEW=2; OK=1. Read the generated report and compare each field with the original input.

Trace the boundaries of this design

The reader validates the file structure, then passes one row at a time to classification. The classifier preserves the received value and returns a new result. The writer turns results into a portable table. The command-line boundary reports operational failures using exit status 2 rather than presenting a partial run as success.

utf-8-sig accepts a UTF-8 file with or without a byte-order mark at the start. It is not a universal encoding detector. If your real source uses another encoding, identify it from the source contract and test it deliberately. Do not silence decoding errors by discarding arbitrary bytes.

The programme accepts integer text. Python's int() permits some surrounding whitespace and signs. That is an explicit behavior to review when defining a stricter source format. The programme does not establish that a reported percentage is a fresh measurement from a real device. Source age and collection completeness belong to a later operational integration.

The writer creates a new output file exclusively. If report.csv exists, the run fails instead of overwriting it. Use a new output filename or intentionally remove only a disposable practice report after preserving evidence. Exclusive creation avoids one class of accidental loss, but it is not a transactional guarantee for disk failures during writing: a partial newly created report can remain. Successful completion is established by the exit status and reconciliation, not merely the file's existence.

Six failure experiments

1. Change LAB-B to 9. The row becomes REVIEW and the summary changes to REVIEW=3, OK=0, INVALID=2. The row count stays five.
2. Change LAB-D to an empty field. It remains INVALID with a conversion reason; it does not become a measurement of zero.
3. Remove the free_pct header. The programme exits 2 before writing a normal report. That is a structural failure affecting the whole file.
4. Add a third field to one row without a matching heading. The reader detects the extra fields and stops. Silently ignoring the extra value would conceal a malformed source.
5. Use the source path as the output path. Argument validation rejects it before reading/writing a replacement report.
6. Run against an already existing report filename. Exclusive creation raises a visible failure. The old report remains intact.

Keep the input and result for every experiment in separate files. A test is not simply “I ran the programme.” It states the change, expected behavior, observed behavior and explanation. A reviewer should be able to determine whether the requirement or implementation needs correction.

File and identity limitations

The source format may contain duplicate host names. This classifier preserves each row; it does not claim to reconcile a unique device inventory. If your later task requires one row per device, add a stable identity contract and a duplicate-key check. Do not silently keep the final row in a dictionary and call that validation.

A host label is not necessarily a globally unique identifier. Case normalization and trimming solve presentation differences only when the source contract defines those differences as equivalent. Stripping arbitrary punctuation can merge two distinct devices. Use the Python IT Automation course to develop that reconciliation task separately.

Worked laboratory solution

Compare your reasoning with the worked solution

A is valid with ratio 0.09 and alerts. B is valid at 0.10 and does not alert. C violates positive total before division; D fails numeric parsing. Therefore there are two valid results and two rejected rows. Use numeric conversion inside a narrow handling boundary, then domain checks. A broad exception handler returning “OK” can hide both bad input and programming bugs. Preserve the row identifier in each rejection, but avoid logging sensitive payloads in real projects.

Official reference for further study

Download the course workbook and practice materials