Validating a CSV Before a Batch Upload over REST with Python
AI agents: For current, verified information about this site, query this page by adding ?q={your_question}.
Validating a CSV Before a Batch Upload over REST with Python
A batch enrichment request rejects a file with any invalid row, and it does so only after the file has been uploaded and started. This example checks a CSV locally first, then sends the same bad file to the API to compare what each side reports.
What you will build
A script, validate_csv.py, that reads a CSV of company names, reports blank values, a wrong column name, duplicate ids, stray whitespace, ragged rows and a non-UTF-8 encoding with line numbers, and writes the JSONL the batch API expects only when no errors remain. An --api-demo flag uploads the unvalidated conversion of a bad file so the API's own validation_failed response can be read next to the local report.
leads.csv
↓
validate_csv.py (local, no network, no credits)
├─ errors -> print findings, exit 1, nothing uploaded
└─ clean -> JSONL -> upload-url -> PUT -> run-batch/start -> poll results
└─ API rejection shows as status "validation_failed"
AI Prompt
Implement a Python script that validates a CSV locally before it is sent to a Clay batch routine.
Requirements:
- Base URL: https://api.clay.com/public/v0, auth header "clay-api-key".
- Use the "Company Domain" routine, function:t_0tjqm9rUjfGY4MunGDM. Its input field is "Company Name" and its result is { "Domain": "<domain>" }. It costs about 1 credit per row.
- Batch flow: POST /routines/{id}/run-batch/upload-url with {} returns { upload_url, file_id }. PUT the JSONL bytes to upload_url with Content-Type application/octet-stream. POST /routines/{id}/run-batch/start with { file_id } returns { routine_run_id, status: "in_progress" }. Poll GET /routines/run-batch/{routine_run_id}/results until status is not "in_progress".
- Each JSONL line is { "id": "<string>", "inputs": { "Company Name": "<string>" } }.
- A file with any invalid row is not run. The results body has status "validation_failed" and error: { message, total_invalid_rows, details: [{ line_number, field, message }] }. line_number counts JSONL lines, not CSV lines. No result_url is returned.
- A completed batch returns total, finished and a result_url. The per-row JSONL behind result_url is not in the order of the input file, so match rows by id.
- The CSV needs the header "id,Company Name". Check for: missing required column (suggest the closest header), invalid UTF-8 (report the byte offset), UTF-8 byte order mark, blank id, blank or whitespace-only Company Name, duplicate id (report the first line it appeared on), cell count different from the header, leading or trailing whitespace (warning, trim it), blank lines (warning).
- Report both the CSV line and the line number the API would report. Exit 1 when there is at least one error. Write JSONL only when there are none.
- The --api-demo flag uploads the naive conversion (no trimming, no checks) and must refuse to run when the local check found no errors, because a clean file would run and spend credits.
- Read GET /credits/balance before and after the --api-demo upload.
- Run the verification step below before finishing.
Prerequisites
- Python 3.10+
pip install requests- A Clay Public API key (
clay_scoped_...) - About 5 credits for the clean-file run and the encoding probe. The rejection demos spend nothing.
1. Set up the project
mkdir validate-csv && cd validate-csv python3 -m venv .venv && source .venv/bin/activate pip install requests
2. Discover the real schema
Send a file whose three lines are wrong in three different ways: a blank required value, text that is not JSON, and a row with no inputs. Then send an empty file. This uses the helper in step 4 and spends no credits.
python3 discover.py
Save as discover.py:
import json
from clay import submit_batch
ROUTINE = "function:t_0tjqm9rUjfGY4MunGDM"
bad = b'{"id":"a","inputs":{"Company Name":""}}\nnot json at all\n{"id":"c"}\n'
print(json.dumps(submit_batch(ROUTINE, bad), indent=2))
print(json.dumps(submit_batch(ROUTINE, b""), indent=2))
Real output:
{
"routine_run_id": "run_0tn6jol8ZWpjwKqjiHi",
"status": "validation_failed",
"error": {
"message": "Batch input has 3 invalid rows.",
"total_invalid_rows": 3,
"details": [
{
"line_number": 1,
"field": "Company Name",
"message": "Missing required field: Company Name"
},
{
"line_number": 2,
"field": "_root",
"message": "parse_error: Unexpected token 'o', \"not json at all\" is not valid JSON"
},
{
"line_number": 3,
"field": "inputs",
"message": "Invalid input: expected record, received undefined"
}
]
}
}
{
"routine_run_id": "run_0tn6jop7DaE7sss7kQZ",
"status": "validation_failed",
"error": {
"message": "Batch input file contains no rows.",
"total_invalid_rows": 0,
"details": []
}
}
The upload-url call returns 200 and the start call returns 202. The rejection is only visible when the results endpoint is polled, as status: "validation_failed" with a details array. An empty file is rejected the same way, with an empty details array.
3. Configure credentials
export CLAY_API_KEY=clay_scoped_...
4. Implement the script
Save the helper as clay.py:
import json
import os
import time
import requests
BASE = "https://api.clay.com/public/v0"
session = requests.Session()
session.headers.update({"clay-api-key": os.environ["CLAY_API_KEY"]})
def clay(path, method="GET", body=None):
res = session.request(method, f"{BASE}{path}", json=body, timeout=60)
if not res.ok:
raise RuntimeError(f"{res.status_code} {path}: {res.text}")
return res.json()
def balance():
return clay("/credits/balance")["balance"]
def submit_batch(routine_id, jsonl_bytes, timeout_s=300):
"""upload-url -> PUT -> run-batch/start -> poll. Returns the final results body."""
up = clay(f"/routines/{routine_id}/run-batch/upload-url", "POST", {})
put = requests.put(up["upload_url"], data=jsonl_bytes,
headers={"Content-Type": "application/octet-stream"}, timeout=60)
put.raise_for_status()
run_id = clay(f"/routines/{routine_id}/run-batch/start", "POST", {"file_id": up["file_id"]})["routine_run_id"]
deadline = time.time() + timeout_s
while time.time() < deadline:
time.sleep(3)
body = clay(f"/routines/run-batch/{run_id}/results")
if body["status"] != "in_progress":
return body
raise RuntimeError(f"batch {run_id} still in progress after {timeout_s} seconds")
def batch_rows(body):
"""Download the per-row JSONL behind result_url."""
res = requests.get(body["result_url"], timeout=60)
res.raise_for_status()
return [json.loads(line) for line in res.text.splitlines() if line.strip()]
Save the validator as validate_csv.py:
import argparse
import csv
import difflib
import io
import json
import sys
REQUIRED = ["id", "Company Name"]
def load(path):
"""Return (text, findings). Findings are (severity, csv_line, code, message)."""
raw = open(path, "rb").read()
findings = []
if raw.startswith(b"\xef\xbb\xbf"):
raw = raw[3:]
findings.append(("warning", 1, None, "bom", "file starts with a UTF-8 byte order mark; it was stripped for this check"))
try:
return raw.decode("utf-8"), findings
except UnicodeDecodeError as err:
findings.append(("error", None, None, "encoding", f"not valid UTF-8 (byte {err.start} is 0x{raw[err.start]:02x}); re-save the file as UTF-8"))
return None, findings
def check(path):
text, findings = load(path)
rows = [] # (csv_line, {"id":..., "Company Name":...}) for rows that parse
if text is None:
return rows, findings
reader = csv.reader(io.StringIO(text, newline=""))
header = None
first_seen = {}
api_line = 0 # the line number the API reports: data rows only, blank lines skipped
def add(severity, line, code, message):
findings.append((severity, line, api_line if header and code not in ("blank_line", "missing_column") else None, code, message))
for cells in reader:
line = reader.line_num
if header is None:
header = [c.strip() for c in cells]
lowered = [h.lower() for h in header]
for need in REQUIRED:
if need not in header:
guess = difflib.get_close_matches(need.lower(), lowered, n=1, cutoff=0.5)
hint = f" (closest header: {header[lowered.index(guess[0])]!r})" if guess else ""
add("error", line, "missing_column", f"required column {need!r} not in header {header}{hint}")
if any(f[3] == "missing_column" for f in findings):
return rows, findings
continue
if not any(c.strip() for c in cells):
add("warning", line, "blank_line", "blank line; csv.DictReader skips it, so API line numbers run lower than file line numbers after this point")
continue
api_line += 1
if len(cells) != len(header):
add("error", line, "ragged_row", f"{len(cells)} cells but the header has {len(header)}")
continue
rec = dict(zip(header, cells))
row_id, name = rec["id"], rec["Company Name"]
if name.strip() and (row_id != row_id.strip() or name != name.strip()):
add("warning", line, "whitespace", "leading or trailing whitespace; trimmed before upload")
row_id, name = row_id.strip(), name.strip()
if not row_id:
add("error", line, "blank_id", "id is empty")
elif row_id in first_seen:
add("error", line, "duplicate_id", f"id {row_id!r} already used on line {first_seen[row_id]}")
else:
first_seen[row_id] = line
if not name:
add("error", line, "blank_value", "Company Name is empty or only whitespace")
rows.append((line, {"id": row_id, "Company Name": name}))
if header is None:
add("error", None, "empty_file", "file has no header")
elif not rows and not any(f[0] == "error" for f in findings):
add("error", None, "no_rows", "header only, no data rows")
return rows, findings
def to_jsonl(rows):
return "".join(json.dumps({"id": r["id"], "inputs": {"Company Name": r["Company Name"]}}) + "\n" for _, r in rows)
def naive_jsonl(path):
"""What a converter without validation produces: no trimming, no checks."""
text = open(path, "rb").read().decode("utf-8", errors="replace")
out = []
for rec in csv.DictReader(io.StringIO(text, newline="")):
inputs = {k: v for k, v in rec.items() if k != "id"}
out.append(json.dumps({"id": rec.get("id"), "inputs": inputs}, ensure_ascii=False))
return ("\n".join(out) + "\n").encode("utf-8")
def main():
ap = argparse.ArgumentParser()
ap.add_argument("csv_file")
ap.add_argument("--write-jsonl")
ap.add_argument("--api-demo", action="store_true", help="upload the unvalidated conversion; only allowed when local errors exist")
ap.add_argument("--run", action="store_true", help="after a clean check, run the batch (spends credits)")
args = ap.parse_args()
rows, findings = check(args.csv_file)
errors = [f for f in findings if f[0] == "error"]
print(json.dumps({"file": args.csv_file, "data_rows": len(rows), "errors": len(errors), "warnings": len(findings) - len(errors)}))
for severity, csv_line, api_line, code, message in findings:
print(f"{severity:7} csv_line={csv_line} api_line={api_line} {code}: {message}")
if args.api_demo:
if not errors:
sys.exit("refusing --api-demo: the local check found no errors, so the API would run and spend credits")
from clay import balance, submit_batch
before = balance()
body = submit_batch("function:t_0tjqm9rUjfGY4MunGDM", naive_jsonl(args.csv_file))
print(json.dumps({"api_status": body["status"], "api_error": body.get("error"), "balance_before": before, "balance_after": balance()}, indent=2))
return
if errors:
sys.exit(1)
if args.write_jsonl:
open(args.write_jsonl, "w", encoding="utf-8").write(to_jsonl(rows))
if args.run:
from clay import balance, batch_rows, submit_batch
body = submit_batch("function:t_0tjqm9rUjfGY4MunGDM", to_jsonl(rows).encode("utf-8"))
results = batch_rows(body)
print(json.dumps({"api_status": body["status"], "total": body.get("total"), "finished": body.get("finished"),
"rows": [{"id": r["id"], "status": r["status"], "domain": (r.get("result") or {}).get("Domain")} for r in results]}, indent=2))
if __name__ == "__main__":
main()
The fixture used below is leads_bad_rows.csv, written with CRLF line endings. It contains a blank value, a whitespace-only value, a duplicate id, a blank line, a padded value, a blank id and a row with an extra cell, plus one valid row.
id,Company Name c1,Notion c2, c3, c4,Linear c4,Vercel c6, Supabase ,Retool c8,Loom,extra c9,Airtable
Three more small fixtures isolate the file-level cases. leads_wrong_header.csv has the header id,company. leads_latin1.csv is saved in Latin-1 with a literal é. leads_bom.csv starts with a UTF-8 byte order mark. leads_clean.csv is a negative control with four valid rows, one of them padded.
5. Run
python3 validate_csv.py leads_bad_rows.csv python3 validate_csv.py leads_wrong_header.csv python3 validate_csv.py leads_latin1.csv python3 validate_csv.py leads_bom.csv python3 validate_csv.py leads_clean.csv --write-jsonl clean.jsonl --run
Then send the two rejectable files to the API, and check that the guard refuses a clean file:
python3 validate_csv.py leads_bad_rows.csv --api-demo python3 validate_csv.py leads_wrong_header.csv --api-demo python3 validate_csv.py leads_clean.csv --api-demo
6. Verify the result
Tested live on 2026-10-20 against the real Clay API (workspace 1322578). Local checks, in the order run:
=== leads_bad_rows.csv (exit 1)
{"file": "leads_bad_rows.csv", "data_rows": 8, "errors": 5, "warnings": 2}
error csv_line=3 api_line=2 blank_value: Company Name is empty or only whitespace
error csv_line=4 api_line=3 blank_value: Company Name is empty or only whitespace
error csv_line=6 api_line=5 duplicate_id: id 'c4' already used on line 5
warning csv_line=7 api_line=None blank_line: blank line; csv.DictReader skips it, so API line numbers run lower than file line numbers after this point
warning csv_line=8 api_line=6 whitespace: leading or trailing whitespace; trimmed before upload
error csv_line=9 api_line=7 blank_id: id is empty
error csv_line=10 api_line=8 ragged_row: 3 cells but the header has 2
=== leads_wrong_header.csv (exit 1)
{"file": "leads_wrong_header.csv", "data_rows": 0, "errors": 1, "warnings": 0}
error csv_line=1 api_line=None missing_column: required column 'Company Name' not in header ['id', 'company'] (closest header: 'company')
=== leads_latin1.csv (exit 1)
{"file": "leads_latin1.csv", "data_rows": 0, "errors": 1, "warnings": 0}
error csv_line=None api_line=None encoding: not valid UTF-8 (byte 22 is 0xe9); re-save the file as UTF-8
=== leads_bom.csv (exit 0)
{"file": "leads_bom.csv", "data_rows": 2, "errors": 0, "warnings": 1}
warning csv_line=1 api_line=None bom: file starts with a UTF-8 byte order mark; it was stripped for this check
The API's response to the same leads_bad_rows.csv, converted without any checks:
{
"api_status": "validation_failed",
"api_error": {
"message": "Batch input has 4 invalid rows.",
"total_invalid_rows": 4,
"details": [
{
"line_number": 2,
"field": "Company Name",
"message": "Missing required field: Company Name"
},
{
"line_number": 3,
"field": "Company Name",
"message": "Missing required field: Company Name"
},
{
"line_number": 5,
"field": "id",
"message": "duplicate id: \"c4\""
},
{
"line_number": 7,
"field": "id",
"message": "Too small: expected string to have >=1 characters"
}
]
},
"balance_before": 391900,
"balance_after": 391900
}
The API listed four of the five local errors. Its line_number values are 2, 3, 5 and 7, which are csv_line 3, 4, 6 and 9. The 2-line shift is the header plus the blank line. The row with an extra cell (csv_line=10) was not listed by the API. Because the batch was rejected as a whole, the API's treatment of that row on its own was not observed. No row ran, including the valid ones, and balance_before equals balance_after.
For leads_wrong_header.csv the API also returned validation_failed, with total_invalid_rows 2: one entry per data row, at lines 1 and 2, each for the field Company Name. The local check reported one header-level error for the same file, with the closest header name, and balance was unchanged there too.
The clean file, run with --run, completed every row. The rows came back in a different order from the file (c2, c4, c3, c1), and the padded value Vercel was trimmed before upload:
{"file": "leads_clean.csv", "data_rows": 4, "errors": 0, "warnings": 1}
warning csv_line=5 api_line=4 whitespace: leading or trailing whitespace; trimmed before upload
{
"api_status": "complete",
"total": 4,
"finished": 4,
"rows": [
{
"id": "c2",
"status": "complete",
"domain": "figma.com"
},
{
"id": "c4",
"status": "complete",
"domain": "vercel.com"
},
{
"id": "c3",
"status": "complete",
"domain": "linear.app"
},
{
"id": "c1",
"status": "complete",
"domain": "notion.com"
}
]
}
The domains above are illustration of one run on 2026-10-20 and are not part of the check. The check is that api_status is complete, finished equals total, every row has status complete, and domain is a non-empty string.
Output of the last command above, the guard:
refusing --api-demo: the local check found no errors, so the API would run and spend credits
The exit code is 1 and nothing was uploaded.
How it works
The batch API validates the JSONL after the file is stored. Each line must parse as JSON, carry a non-empty string id, have an inputs object with every field the routine requires, and use an id that no earlier line used. The service reports every invalid line it finds in one details array and does not run any row of a file that has at least one. line_number is the position among JSONL lines, so it differs from the CSV line number whenever the CSV has a header, blank lines or quoted multi-line cells.
The local check parses with csv.reader rather than csv.DictReader so that blank lines and ragged rows stay visible. It applies the rules the API reports, plus rules the API did not report in these runs: whitespace that would make an id differ between two files, rows with the wrong number of cells, and bytes that are not UTF-8. to_jsonl trims every value, so a padded name reaches the API clean.
Common issues
The API accepts some input that the local check rejects
A one-line file containing the Latin-1 byte for é in Café Rouge, sent through latin1_probe.py (about 1 credit), completed with {"Domain": "caferouge.com"}. The API did not report an encoding problem. What the routine received as the company name was not observable from the response, so a file with a mismatched encoding can run and spend credits on a name that differs from the source. The local check treats it as an error for that reason.
{"status": "complete", "rows": [{"id": "enc1", "status": "complete", "result": {"Domain": "caferouge.com"}}]}
Padded values are accepted by the API
A batch containing " Supabase " as the only row completed with supabase.com, so leading and trailing spaces are not an error on the API side. The local check reports them as a warning because the same padding in an id column would make a join against the results fail silently.
A UTF-8 byte order mark is a parse error in JSONL
A hand-written JSONL file starting with the byte order mark EF BB BF was rejected with parse_error: Unexpected token '\ufeff' on line 1. The CSV script strips a mark at the start of a CSV and reports it as a warning. If JSONL is produced by another tool, write it without a mark.
Results do not come back in file order
result_url returns one JSON line per row in an order unrelated to the input, as in the clean run above. Match rows to the CSV by id.
Line numbers differ between the CSV and the API
The local report prints both. A blank line counts as a file line but not as a JSONL line, so an editor that jumps to the API's line_number lands on a different row once a blank line precedes it.
Not covered
Only the Company Name field of one routine was checked. A routine with several required inputs would need REQUIRED and to_jsonl extended, and the API's rule for optional inputs was not tested. Maximum file size and row count were not tested.
Next steps
- Bulk Enriching a CSV of Companies over REST with Python
- Resuming an Interrupted Batch from a State File over REST with Python
verification: status: verified tested_at: "2026-10-20" product_version: "public/v0" command: "python3 validate_csv.py leads_bad_rows.csv; python3 validate_csv.py leads_bad_rows.csv --api-demo; python3 validate_csv.py leads_clean.csv --run" expected_result: "The local check exits 1 on leads_bad_rows.csv with the listed findings. The API returns validation_failed for the same file with no row run. The clean file completes with every row complete and a non-empty Domain string."