Parse records before calculating totals
Python's standard library can parse and summarize a small CSV without pandas. This recipe validates an expense ledger, groups spending by category and returns integer-cent totals as JSON. It handles a comma inside Support, EU and a quoted newline inside Research Labs. Splitting on commas or physical lines would break those records.
The parser accepts one fixed contract: comma-delimited UTF-8 text, the four headers below in that order, EUR amounts and unique expense IDs. Any invalid nonblank record rejects the entire dataset. No partial aggregate is printed. Blank physical lines outside quoted fields are ignored by the CSV reader; a header without expense rows is rejected.
Scroll horizontally to see every column.
| Field | Rule |
|---|---|
| expense_id | Unique EXP- followed by exactly three ASCII digits |
| category | Nonempty after whitespace normalization; at most 64 characters |
| amount_cents | Unsigned decimal integer from 0 to 1000000; no decimal point or leading zeroes |
| currency | Exactly EUR; mixing currencies is rejected |
Use a fixture with quoted commas and newlines
expense_id,category,amount_cents,currencyEXP-001,"Support, EU",1250,EUREXP-002,Cloud,4800,EUREXP-003,"Support, EU",750,EUREXP-004,"ResearchLabs",2000,EURThis fixture has four expense records but one record spans two physical lines. The script uses csv.DictReader over io.StringIO(..., newline=""), checks the exact header list and rejects missing or extra fields. With strict=True, parser-detected syntax errors reject the dataset before a summary is printed. The recipe uses the reviewed comma-and-double-quote dialect shown here. Python 3.13 csv reference, StringIO reference.
These are synthetic expenses, not customer records. When adapting the recipe, keep the dataset small and reviewed. Its input limit is 16 KiB, its row limit is 200 and its field limit is 1,024 characters. The complete source submitted to cpuOS must also stay within the API's separate 64 KiB code limit.
Run the complete Python program locally
import csvimport ioimport jsonimport reimport sysCSV_TEXT = "expense_id,category,amount_cents,currency\nEXP-001,\"Support, EU\",1250,EUR\nEXP-002,Cloud,4800,EUR\nEXP-003,\"Support, EU\",750,EUR\nEXP-004,\"Research\nLabs\",2000,EUR\n"HEADERS = ["expense_id", "category", "amount_cents", "currency"]MAX_INPUT_BYTES = 16 * 1024MAX_ROWS = 200def analyze(text): if not isinstance(text, str) or not text or "\0" in text: raise ValueError("CSV input must be nonempty text without NUL characters.") if len(text.encode("utf-8")) > MAX_INPUT_BYTES: raise ValueError("CSV input exceeds the recipe's 16 KiB limit.") csv.field_size_limit(1024) reader = csv.DictReader(io.StringIO(text, newline=""), strict=True) if reader.fieldnames != HEADERS: raise ValueError("CSV headers must exactly match the documented order.") groups = {} seen_ids = set() count = 0 total = 0 for row in reader: line = reader.line_num if None in row or any(row.get(field) is None for field in HEADERS): raise ValueError("Wrong field count at CSV line " + str(line) + ".") expense_id = row["expense_id"] if not re.fullmatch(r"EXP-[0-9]{3}", expense_id) or expense_id in seen_ids: raise ValueError("Invalid or duplicate expense ID at CSV line " + str(line) + ".") category = " ".join(row["category"].split()) if not category or len(category) > 64 or any(ord(c) < 32 for c in category): raise ValueError("Invalid category at CSV line " + str(line) + ".") amount_text = row["amount_cents"] if not re.fullmatch(r"0|[1-9][0-9]{0,6}", amount_text): raise ValueError("Amount must be unsigned integer cents at CSV line " + str(line) + ".") amount = int(amount_text) if amount > 1_000_000 or row["currency"] != "EUR": raise ValueError("Amount or currency is outside this recipe's contract at CSV line " + str(line) + ".") count += 1 if count > MAX_ROWS: raise ValueError("CSV input exceeds the recipe's 200-row limit.") seen_ids.add(expense_id) group = groups.setdefault(category, {"row_count": 0, "total_cents": 0}) group["row_count"] += 1 group["total_cents"] += amount total += amount if count == 0: raise ValueError("CSV input must contain at least one expense row.") return { "currency": "EUR", "row_count": count, "total_cents": total, "categories": [dict(category=name, **groups[name]) for name in sorted(groups)], }if __name__ == "__main__": try: print(json.dumps(analyze(CSV_TEXT), ensure_ascii=False, sort_keys=True)) except csv.Error: print("CSV syntax invalid.", file=sys.stderr) sys.exit(1) except ValueError as error: print(str(error), file=sys.stderr) sys.exit(1)python3 --versionpython3 analyze-csv.pyOn success, stdout contains one JSON value. Missing fields, duplicates, invalid amounts and wrong currencies raise fixed diagnostic messages without echoing the row's values. The program prints diagnostics to stderr and exits with code 1. It does not use a skipped-row count to disguise an incomplete ledger.
Check the result against known category totals
{ "currency": "EUR", "row_count": 4, "total_cents": 8800, "categories": [ { "category": "Cloud", "row_count": 1, "total_cents": 4800 }, { "category": "Research Labs", "row_count": 1, "total_cents": 2000 }, { "category": "Support, EU", "row_count": 2, "total_cents": 2000 } ]}The two Support records total 2,000 cents; Cloud contributes 4,800 and Research Labs contributes 2,000. Their sum is 8,800 cents across four rows. Categories are sorted for stable output. Whitespace normalization joins the quoted newline into a single space, while the comma remains part of the category name.
The program produces a JSON value with Python's json module. Your application should parse that value and check the row count, currency and sum of category totals. Integer cents avoid binary floating-point rounding for this contract; taxes, currency conversion and arbitrary decimal precision need their own rules.
Save this fixture verifier as verify-result.mjs. It compares the entire parsed object in result.json with the expected result and reports a fixed diagnostic on failure. For a real ledger, replace fixture equality with your versioned contract and appropriate totals checks.
import assert from "node:assert/strict"import { readFileSync } from "node:fs"const expected = { "currency": "EUR", "row_count": 4, "total_cents": 8800, "categories": [ { "category": "Cloud", "row_count": 1, "total_cents": 4800 }, { "category": "Research Labs", "row_count": 1, "total_cents": 2000 }, { "category": "Support, EU", "row_count": 2, "total_cents": 2000 } ]}try { const actual = JSON.parse(readFileSync("result.json", "utf8")) assert.deepStrictEqual(actual, expected) console.log("Fixture result matches.")} catch { // Assertion and parsing errors can contain data; keep diagnostics fixed. console.error("Fixture result did not match the expected contract.") process.exitCode = 1}Submit the same source as a Python job
Follow the quickstart, connect an online Docker worker and create a workspace API key. Save the published Node client from JavaScript code execution as run-job.mjs in the same local directory as analyze-csv.py. The client reads that source file on your machine and submits its text with template: "python"; it does not upload a CSV file to the worker.
# CPUOS_API_KEY is already set in your trusted client environment.# Generate a key once for this intended execution and retain it for recovery.export CPUOS_IDEMPOTENCY_KEY="$(node -p 'crypto.randomUUID()')"node run-job.mjs python analyze-csv.py > result.json && node verify-result.mjsThe client requests 1 CPU, 256 MiB and a 30-second job timeout. It polls until terminal state and prints parsed stdout only after completed, exit code 0, no execution error and outputTruncated: false. Compare the returned object with the expected fixture result. Retain the same idempotency key and source after an uncertain submission; generating a fresh key requests another execution.
cpuOS currently runs trusted team code in restricted Docker containers on your worker. Job code and output pass through its EU-hosted control plane; the worker location is your choice. Jobs have no network, package installation, file-transfer API or persistent session. Keep the API key in the submitting application, outside job source.
Change the business rules explicitly
- Test a quoted comma, a quoted newline, a missing field and a malformed quote before adapting an export format.
- Define a separate contract for another delimiter, header order or currency. Do not guess column types from a few rows.
- Keep the all-or-nothing behavior for this ledger. If an application intentionally quarantines rows, return an explicit incomplete status and approved error metadata.
- Keep stdout small. Truncated JSON is an incomplete result, even when the process exits successfully.
Use n8n HTTP jobs for an approved workflow step or LangChain tool wiring for a controlled calculation tool. gpuOS open models can propose a task; your application still authorizes the input and checks the execution result. For lifecycle errors, retries and cancellation, use the Python API guide.