Filesystem & Structured Data Pipelines.

Lesson 01 taught you to build a local project and Lesson 02 taught you to make it configurable from a command line. This lesson connects those skills to real filesystem operations and structured data. Our data is fictional, but the Python parsing and validation are real.

• Structure a file pipeline with explicit location, reading, parsing, validation, transformation and publication stages.
• Use pathlib to reason about relative paths, root boundaries and local file access without relying on text prefixes.
• Specify UTF-8 encodings, recognize BOM-bearing CSV exports and open CSV files with newline=''.
• Read and validate CSV headers and rows with DictReader, including quotes, extra fields and missing values.
• Normalize only known status values and report malformed, duplicate or unknown records separately.
• Parse and serialize stable JSON while validating schema, Unicode behavior and nonstandard numeric values.
• Bound file inputs and distinguish missing, unreadable and oversized files from valid empty exports.
• Stage output in the target folder and replace existing reports only after validation and an explicit write decision.
• Combine a CLI from Lesson 02 with an offline file pipeline that can be tested safely in both browser and local environments.

Prerequisite: Complete Command-Line Tools & Configuration or know how to use argparse and run a local Python script. If CSV, JSON, exceptions or functions still feel unfamiliar, review the Python Fundamentals path before continuing.

A working file pipeline is more than calling open() and writing a result. A reliable tool has a clear sequence: locate the requested file, read it with an intentional encoding, parse its format, validate what the data means, transform accepted rows, then publish a report. Each stage should make its own result explicit so an invalid source cannot turn into a successful-looking empty report. Use fictional inventory exports first; never aim beginner scripts at production shares or customer data.

def summarize(rows):
    offline = sum(1 for row in rows if row['status'] == 'offline')
    return {'total': len(rows), 'offline': offline}
print(summarize([{'status':'offline'},{'status':'online'}]))

host,status,site
SW-01,online,HQ
AP-02,offline,BR1
AP-02,online,BR1
FW-01,offline,"HQ, East"
BAD,maintenance,HQ

pathlib.Path represents filesystem paths and works across operating systems. A relative path such as samples/inventory.csv is usually resolved against the process working directory, which can differ when launched from an editor, Task Scheduler or a service. Use a documented project root or explicit absolute input path rather than assuming the current directory. If a tool accepts a user-supplied filename under a known data folder, verify the resolved path stays inside that folder. Be aware that resolving paths and then opening them isn't a complete defense against a malicious actor changing symlinks concurrently; for this learning tool we use only trusted local files.

from pathlib import Path
base = Path('samples').resolve()
candidate = (base / 'inventory.csv').resolve()
allowed = candidate.is_relative_to(base)
print(allowed)

from pathlib import Path

base = Path('/northstar/incoming').resolve()
name = Path('exports/inventory.csv')
allowed = (not name.is_absolute() and '..' not in name.parts)
target = (base / name).resolve()
print(allowed and target != base and target.is_relative_to(base))

File contents start as bytes. When Python reads text, it decodes those bytes with an encoding. Specify encoding='utf-8' rather than relying on the machine's default, and use 'utf-8-sig' when a CSV exporter may include a UTF-8 byte-order mark: that codec removes a leading BOM while also reading normal UTF-8. A wrong encoding may raise UnicodeDecodeError or produce damaged text. For CSV, open(path, newline='', encoding='utf-8-sig') so Python's csv module can handle record newlines correctly, including quoted fields containing line breaks. Do not solve encoding errors by blindly ignoring undecodable bytes.

from io import StringIO
import csv
sample = '\ufeffhost,status\nSW-01,online\n'
with StringIO(sample.lstrip('\ufeff'), newline='') as stream:
    reader = csv.DictReader(stream)
    print(reader.fieldnames)

from pathlib import Path
from tempfile import TemporaryDirectory

with TemporaryDirectory() as folder:
    path = Path(folder) / 'inventory.csv'
    path.write_bytes(b'\xef\xbb\xbfhost,status\nSW-01,online\n')
    with open(path, 'r', encoding='utf-8-sig', newline='') as stream:
        text = stream.read()
    print(text.splitlines()[0])

Real CSV can include commas, quotes and even newlines inside quoted fields. csv.DictReader maps each row to column names, which makes the logic independent of a particular column order. It still needs schema validation: fieldnames can be missing or duplicated, and extra columns appear under a None key by default while short rows get None values. Treat suspicious row shapes as invalid evidence, not values to guess. When exporting with DictWriter, declare the output field order and use writeheader(); a stable schema helps downstream tools and spreadsheets.

import csv
from io import StringIO
csv_text = 'status,host,site\noffline,SW-01,"HQ, East"\n'
rows = list(csv.DictReader(StringIO(csv_text, newline='')))
print(rows[0]['site'])

Parsing turns bytes into fields, but that isn't the same as proving those fields are usable. Trim whitespace around host and status, normalize known status values, and define what to do with blank, duplicated or unexpected records. The example treats online and offline as known statuses; maintenance, missing and misspelled statuses are not automatically online. For deterministic reporting, sort accepted hosts explicitly. If duplicates are ambiguous, reject or flag them instead of silently preferring the first record. Return structured validation details so a technician can tell clean exports from partially rejected data.

allowed={'online','offline'}
raw = {'host':' AP-02 ', 'status':' OFFLINE '}
host=raw['host'].strip()
status=raw['status'].strip().lower()
print(host, status, status in allowed)

from collections import Counter

hosts = ['SW-01', 'AP-02', 'sw-01']
normalized = [host.upper() for host in hosts]
frequency = Counter(normalized)
print(frequency['SW-01'])
print(sorted(host for host, count in frequency.items() if count == 1))

JSON stores objects, arrays, strings, numbers, booleans and null. json.loads() can parse valid JSON text, but a valid JSON document may still have the wrong top-level shape; check that the expected object and collection keys are present. Use json.dumps(report, sort_keys=True, indent=2, ensure_ascii=False) for readable, deterministic output when field ordering and Unicode are important to your project. `ensure_ascii=False` retains Unicode characters in the text, so write the result with UTF-8. Treat NaN and Infinity as invalid reporting numbers rather than relying on Python's permissive default output; allow_nan=False prevents those nonstandard values from being emitted.

import json
report={'site':'São Paulo','offline':1,'total':2}
print(json.dumps(report,sort_keys=True,ensure_ascii=False))

A file that looks like a harmless export can be unexpectedly large or malformed. Before reading an entire file, apply a sensible byte limit for your tool and require a known source. Avoid calling read_text() on unknown multi-gigabyte exports, and remember size checks are not race-proof on files that other processes can alter. A first small utility can read bounded inputs entirely; later large-scale pipelines can iterate rows instead of building a huge list. Design errors so a too-large or unreadable export is rejected, not reported as a clean inventory with zero devices.

from pathlib import Path
path=Path('samples/inventory.csv')
max_bytes=1024*1024
if path.is_file() and path.stat().st_size<=max_bytes:
    print('Size is acceptable to inspect')
else:
    print('Missing or outside size limit')

Writing directly over last week's valid report can leave a half-written file if your process stops. A safer local pattern is to build the complete report in memory, write a temporary file in the destination directory, close it, then use os.replace(temp, destination) to replace the old report. On supported local filesystems, replacement is atomic at the filename level; it is not a guarantee of crash durability or universal network-share behavior. os.replace overwrites the destination, so require an approved output location and explicit overwrite policy. Clean up temporary artifacts on failure and keep your original input export unchanged.

import json
payload={'site':'HQ','total':2}
text=json.dumps(payload,sort_keys=True,ensure_ascii=False,allow_nan=False)
print(text)

import os
from pathlib import Path
from tempfile import TemporaryDirectory, NamedTemporaryFile

with TemporaryDirectory() as folder:
    destination = Path(folder) / 'report.json'
    temporary = None
    try:
        with NamedTemporaryFile('w', encoding='utf-8',
                                dir=folder, delete=False) as stream:
            temporary = Path(stream.name)
            stream.write('{"total": 2}\n')
        if destination.exists():
            raise FileExistsError('overwrite not approved')
        os.replace(temporary, destination)
        temporary = None
        print(destination.read_text().strip())
    finally:
        if temporary is not None:
            temporary.unlink(missing_ok=True)

The Applied Python project can now accept --input and --output arguments, find a known local inventory CSV, validate required host and status fields, and produce a deterministic JSON report. Keep parsing and transformations as pure functions where possible; let main(argv) coordinate file operations and return an exit code. Use --dry-run or a preview default to inspect the proposed report without writing. A useful runbook says which files may be touched, what a successful report contains, and what the tool does when an input is absent or malformed. Test missing headers, quoted commas, duplicate hosts, empty inventories, and an existing output report before calling the workflow reliable.

def report_from_rows(rows):
    accepted = [r for r in rows if r['status']=='offline']
    return {'total':len(rows),'offline_hosts':sorted(r['host'] for r in accepted)}
print(report_from_rows([{'host':'AP-02','status':'offline'}]))

Build on the local northstar-python-tool project from Lesson 01. This example uses only the standard library and deliberately reads a fictional CSV in samples/. It previews a JSON report by default, leaving files untouched. An explicit --write saves a report inside reports/; a separate --replace is required if that output already exists.

host,status,site
SW-01,online,HQ
AP-02,offline,BR1
AP-02,online,BR1
FW-01,offline,"HQ, East"
BAD,maintenance,HQ

Save the following complete Python source as inventory_pipeline.py in your project root. Read the main() function and report logic before copying it; the code illustrates the intended design rather than replacing the independent challenges in Practice.

import argparse
import csv
import json
import os
import sys
from collections import Counter
from io import StringIO
from pathlib import Path
from tempfile import NamedTemporaryFile

ROOT = Path(__file__).resolve().parent
MAX_BYTES = 256 * 1024

def within(folder, filename):
    name = Path(filename)
    base = (ROOT / folder).resolve()
    if not filename.strip() or name.is_absolute() or ".." in name.parts:
        raise ValueError("Provide a relative filename inside " + folder)
    target = (base / name).resolve()
    if target == base or not target.is_relative_to(base):
        raise ValueError("Path leaves the allowed " + folder + " directory")
    return target

def read_csv(path):
    with path.open("rb") as stream:
        raw = stream.read(MAX_BYTES + 1)
    if len(raw) > MAX_BYTES:
        raise ValueError("Input exceeds 256 KiB")
    return raw.decode("utf-8-sig")

def analyze(text):
    reader = csv.DictReader(StringIO(text, newline=""), strict=True)
    fields = reader.fieldnames or []
    if len(fields) != len(set(fields)) or not {"host","status"} <= set(fields):
        raise ValueError("Missing or repeated host/status columns")
    possible = []
    rejected = 0
    for row in reader:
        host, status = row.get("host"), row.get("status")
        if (None in row or any(value is None for value in row.values())
            or not isinstance(host, str) or not host.strip()
            or not isinstance(status, str)
            or status.strip().lower() not in {"online","offline"}):
            rejected += 1
            continue
        possible.append((host.strip().upper(), status.strip().lower()))
    counts = Counter(host for host, status in possible)
    accepted = [(host, status) for host, status in possible if counts[host] == 1]
    rejected += len(possible) - len(accepted)
    return {
        "accepted": len(accepted),
        "rejected": rejected,
        "offline_hosts": sorted(host for host, status in accepted if status == "offline")
    }

def write_staged(path, report, replace=False):
    if not path.parent.is_dir():
        raise ValueError("Output directory does not exist")
    if path.exists() and not replace:
        raise ValueError("Output exists; use --replace to approve overwrite")
    text = json.dumps(report, sort_keys=True, ensure_ascii=False, allow_nan=False, indent=2) + "\n"
    temporary = None
    try:
        with NamedTemporaryFile(mode="w", encoding="utf-8", dir=path.parent,
                                prefix=".inventory-", suffix=".tmp", delete=False) as stream:
            temporary = Path(stream.name)
            stream.write(text)
        os.replace(temporary, path)
        temporary = None
    finally:
        if temporary is not None:
            temporary.unlink(missing_ok=True)

def main(argv=None):
    parser = argparse.ArgumentParser(description="Fictional, offline inventory file pipeline")
    parser.add_argument("--input", default="inventory.csv", help="Filename within samples/")
    parser.add_argument("--output", default="summary.json", help="Filename within reports/")
    parser.add_argument("--write", action="store_true", help="Publish report to local reports/")
    parser.add_argument("--replace", action="store_true", help="Allow replacing an existing output")
    args = parser.parse_args(argv)
    if args.replace and not args.write:
        parser.error("--replace requires --write")
    try:
        source = within("samples", args.input)
        destination = within("reports", args.output)
        report = analyze(read_csv(source))
        if args.write:
            write_staged(destination, report, replace=args.replace)
            print("Saved report to " + str(destination))
        else:
            print(json.dumps(report, sort_keys=True, ensure_ascii=False, indent=2))
        return 0
    except (ValueError, OSError, UnicodeError, csv.Error) as exc:
        print("Pipeline error: " + str(exc), file=sys.stderr)
        return 1

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

The first exercise checks pathname boundaries. The full coding clinic then explores actual UTF-8 file reads, malformed CSV rows, duplicate hosts, JSON schema checks, deterministic reports and staged writes. The last task makes you design an end-to-end inventory analyzer tested against unfamiliar data.

The official Python documentation is the source of truth for path methods, CSV parsing, JSON serialization and temporary-file behavior. Some filesystem guarantees vary across platforms; read the caveats before adapting local scripts for operational use.