Learn the concept
Build the capstone incrementally. First parse and validate one record. Then process a collection while preserving individual failure outcomes. Add storage only after the transformation behavior is tested. Finally connect the CLI and report writer.
Use parameterized SQL, a unique document ID, and a transaction for each defined atomic unit. Do not leave the database claiming a document was processed if the corresponding required operation failed. If the report is generated separately, describe that boundary and how it can be regenerated.
Test with temporary files and a fresh database. Run the same input twice to check idempotency, then change a source version to test updates. Keep the application usable without network access. This assignment is a build specification, not a hidden dependency on an external course.
Run and inspect
expected_summary = {"accepted": 5, "rejected": 5, "total": 10}
assert expected_summary["accepted"] + expected_summary["rejected"] == expected_summary["total"]
Your exercise
Implement the CLI from the previous lesson. Add unit tests for validation and integration tests for file-to-database processing and repeated runs.
Check your understanding
The summary reconciles all input records, repeated runs do not duplicate data, and failures leave understandable recovery steps.
Guided workshop: a complete offline importer
Save the following as import_documents.py. It reads JSON Lines, validates each record, updates a local SQLite table by ID, and reports rejected inputs without discarding good ones. The sample fixture is built into the script so your first run needs no download or API key.
# academy-check
import json
import sqlite3
from pathlib import Path
from tempfile import TemporaryDirectory
def ingest(source, connection):
accepted = rejected = 0
errors = []
connection.execute("CREATE TABLE IF NOT EXISTS documents (id TEXT PRIMARY KEY, text TEXT NOT NULL)")
with source.open(encoding="utf-8") as lines, connection:
for number, line in enumerate(lines, 1):
try:
record = json.loads(line)
if not isinstance(record, dict):
raise ValueError("object_required")
if not all(isinstance(record.get(k), str) and record[k].strip() for k in ("id", "text")):
raise ValueError("id_and_text_required")
except ValueError as error:
rejected += 1
errors.append({"line": number, "reason": str(error)})
continue
connection.execute(
"INSERT INTO documents VALUES (?, ?) ON CONFLICT(id) DO UPDATE SET text=excluded.text",
(record["id"].strip(), record["text"].strip()),
)
accepted += 1
return {"accepted": accepted, "rejected": rejected, "errors": errors}
with TemporaryDirectory() as folder:
source = Path(folder) / "documents.jsonl"
source.write_text('{"id":"a","text":"hello"}\nnot json\n{"id":"b","text":"world"}\n', encoding="utf-8")
with sqlite3.connect(Path(folder) / "documents.db") as db:
report = ingest(source, db)
assert (report["accepted"], report["rejected"]) == (2, 1)
ingest(source, db)
assert db.execute("SELECT count(*) FROM documents").fetchone()[0] == 2
assert report["errors"][0]["line"] == 2
print("import, rejection, and replay checks passed")
Validation errors are per-record outcomes. Database errors propagate and roll back this batch, so the function never reports a successful import when persistence failed. accepted counts input records accepted, not newly inserted rows; importing the same two records twice leaves two stored documents. This implementation uses last-write-wins updates. Version-aware updates and deletion belong in the ingestion module of the AI path.
Independent challenge: replace the embedded fixture with CLI arguments for source and database paths, add a report file, and supply ten records with five valid and five invalid cases. Test an unreadable source, a numeric ID, empty text, malformed JSON, an array, duplicate IDs, and a changed document. Document your exit codes and update policy.
Solution review: the database transaction cannot atomically commit a separate filesystem report. Commit data first, then generate the report from persisted import outcomes if durable recovery is required. Do not promise atomicity across both systems merely because a with block surrounds the code.
Exit assessment: from a clean folder, another learner can run the importer, explain every rejected line, rerun it without duplicate rows, and change one source record and observe exactly one updated document. Submit your implementation, fixtures, assertions, and an explanation of the transaction boundary before moving on to model APIs.