Capstone: your inventory reporting tool
Build a tested inventory report that keeps good records and explains rejected ones.
Bring the pieces together
Your team receives an inventory CSV and a JSON policy. Build a reusable build_report(inventory_path, policy_path, output_dir) function that returns the summary dictionary and writes three reports. Keep file handling at the edges and give validation and selection clear places in the workflow.
The required input columns are asset_id, hostname, status, memory_gb, and disk_free_gb. Strip identity fields, uppercase asset IDs, and lowercase hostnames and status. IDs and names must be nonempty. Status must be active or inactive. Memory must be a whole number at least 1, and disk free must be a whole number at least 0. Extra CSV cells or missing required values make a row invalid.
Apply a precise policy
The policy must be a JSON object with integer minimum_memory_gb (at least 1) and integer minimum_disk_free_gb (at least 0). Booleans are not accepted as numbers. Missing files, malformed JSON, invalid policy, or missing CSV columns must raise a clear exception before writing reports. Row-level mistakes belong in the error report so other rows can continue.
The first occurrence reserves each nonempty, normalised asset ID, even if that row is invalid. Every later occurrence becomes a duplicate_id error. Other bad rows use invalid_row. Error row numbers start at 2 for the first data record and follow input order. Include a nonempty explanatory message.
Produce a report people can review
Include only valid active assets in inventory_report.csv, sorted by asset ID. Its columns are asset_id,hostname,memory_gb,disk_free_gb,flags. Flag values below the policy minimum with low_memory and low_disk, joined in that order by a semicolon. A record meeting both thresholds has an empty flags field. Equality meets a minimum.
Write errors.csv with columns row,asset_id,error_code,message. Write summary.json with integer counts total_rows, valid_rows, active_rows, flagged_rows, and error_rows. Valid rows include inactive assets; duplicate or invalid rows do not count as valid. Always write headers, even for an empty report, and replace prior output on every successful run.
Use UTF-8 and the CSV library so names containing commas or accents survive. Keep a demonstration behind a main guard. Checks supply normal, empty, invalid, duplicate, missing-file, Unicode, and repeat-run datasets. These are fictional inventory records; your program reads virtual files and does not inspect a real computer.
import csv
import json
from pathlib import Path
policy = json.loads(Path("assets/policy.json").read_text(encoding="utf-8"))
with open("assets/inventory.csv", newline="", encoding="utf-8") as source:
for row in csv.DictReader(source):
print(row["asset_id"], row["hostname"], row["memory_gb"])
print("Policy:", policy)
Put it into practice.
Implement build_report(inventory_path, policy_path, output_dir) to the contracts above. Return the summary and write inventory_report.csv, errors.csv, and summary.json. Use the supplied assets for a guarded demonstration. Keep failures explicit and repeated reports free of duplicates.
See an explained solution
import csv
import json
from pathlib import Path
def load_policy(path):
policy = json.loads(Path(path).read_text(encoding="utf-8"))
if not isinstance(policy, dict):
raise ValueError("Policy must be an object")
for key, minimum in [("minimum_memory_gb", 1), ("minimum_disk_free_gb", 0)]:
value = policy.get(key)
if type(value) is not int or value < minimum:
raise ValueError(f"Invalid policy field: {key}")
return policy
def parse_asset(row):
asset_id = row["asset_id"].strip().upper()
hostname = row["hostname"].strip().lower()
status = row["status"].strip().lower()
if not asset_id or not hostname or status not in {"active", "inactive"}:
raise ValueError("Missing identity or invalid status")
memory = int(row["memory_gb"])
disk = int(row["disk_free_gb"])
if memory < 1 or disk < 0 or None in row:
raise ValueError("Invalid capacity or extra columns")
return {"asset_id": asset_id, "hostname": hostname, "status": status,
"memory_gb": memory, "disk_free_gb": disk}
def write_csv(path, fields, rows):
with Path(path).open("w", newline="", encoding="utf-8") as target:
writer = csv.DictWriter(target, fieldnames=fields)
writer.writeheader()
writer.writerows(rows)
def build_report(inventory_path, policy_path, output_dir):
policy = load_policy(policy_path)
reports, errors, seen = [], [], set()
total_rows, valid_rows = 0, 0
with Path(inventory_path).open(newline="", encoding="utf-8") as source:
reader = csv.DictReader(source)
required = {"asset_id", "hostname", "status", "memory_gb", "disk_free_gb"}
if not required.issubset(reader.fieldnames or []):
raise ValueError("Inventory is missing required columns")
for number, row in enumerate(reader, start=2):
total_rows += 1
asset_id = (row.get("asset_id") or "").strip().upper()
if asset_id and asset_id in seen:
errors.append({"row": number, "asset_id": asset_id,
"error_code": "duplicate_id", "message": "Later duplicate asset ID"})
continue
if asset_id:
seen.add(asset_id)
try:
asset = parse_asset(row)
except (ValueError, TypeError, AttributeError, KeyError) as error:
errors.append({"row": number, "asset_id": asset_id,
"error_code": "invalid_row", "message": str(error)})
continue
valid_rows += 1
if asset["status"] != "active":
continue
flags = []
if asset["memory_gb"] < policy["minimum_memory_gb"]:
flags.append("low_memory")
if asset["disk_free_gb"] < policy["minimum_disk_free_gb"]:
flags.append("low_disk")
reports.append({key: asset[key] for key in ("asset_id", "hostname", "memory_gb", "disk_free_gb")}
| {"flags": ";".join(flags)})
reports.sort(key=lambda row: row["asset_id"])
summary = {"total_rows": total_rows, "valid_rows": valid_rows,
"active_rows": len(reports), "flagged_rows": sum(bool(r["flags"]) for r in reports),
"error_rows": len(errors)}
output = Path(output_dir)
output.mkdir(parents=True, exist_ok=True)
write_csv(output / "inventory_report.csv", ["asset_id", "hostname", "memory_gb", "disk_free_gb", "flags"], reports)
write_csv(output / "errors.csv", ["row", "asset_id", "error_code", "message"], errors)
(output / "summary.json").write_text(json.dumps(summary, indent=2), encoding="utf-8")
return summary
if __name__ == "__main__":
print(json.dumps(build_report("assets/inventory.csv", "assets/policy.json", "output"), indent=2))
The solution validates global inputs first, records row-level errors without losing good data, and separates parsing from policy flags and file writing. Normalised IDs make duplicate handling deterministic. CSV writers preserve quoting, and the summary is built from the same collections written to disk.