Build Data Extraction

From One Extraction to a Pipeline

Lessons 101–108 extracted data from one document at a time. Real projects extract from thousands of documents, and need more than a function call:

• Concurrency, so many documents are processed in reasonable time (Lesson 75).
• Idempotency, so rerunning the job does not redo finished work or create duplicates.
• Storage, so results can be queried and reviewed later.
• Validation and review, so bad records are flagged rather than silently stored.
• Observability, so you can see how many documents succeeded, failed, or were refused.

This lesson builds a small, complete extraction pipeline: event details are extracted from announcements, checked, stored in SQLite, and summarized.

documents ──► skip finished ──► extract (concurrently) ──► check ──► store ──► summary
                 (database)        (structured output)     (rules)   (SQLite)

Module Structure

The pipeline is split into two modules, so the extraction logic can be tested on its own (Lesson 110):

event-extraction/
├── events_extractor.py   # Schema, checks, and one extraction call
└── pipeline.py           # Concurrency, storage, and the run summary

Module 1: events_extractor.py

The extractor defines the schema, the checks, and one function that extracts a single announcement. Every outcome is returned as an Extraction record, including failures, so the pipeline never has to handle exceptions itself.

"""events_extractor.py: extract event details from announcements, with checks."""

from dataclasses import dataclass, field
from datetime import date

import anthropic
from anthropic import AsyncAnthropic
from pydantic import BaseModel, Field, ValidationError

MODEL = "claude-opus-5-5"
SYSTEM = (
    "You extract event details from announcements. Use only what the announcement states. "
    "Use null for any field it does not mention, and never guess a date, venue, or fee. "
    "Write fee_inr as a whole number of rupees, using 0 for free events."
)


class EventInfo(BaseModel):
    title: str | None = Field(description="Event name, or null if not stated")
    date: str | None = Field(description="Event date in YYYY-MM-DD format, or null if not stated")
    venue: str | None = Field(description="Place where the event happens, or null")
    fee_inr: int | None = Field(description="Fee in whole rupees; 0 if free; null if not stated")
    registration_url: str | None = Field(description="Registration link exactly as written, or null")


def check_event(event: EventInfo, source_text: str) -> list[str]:
    problems = []
    if not any([event.title, event.date, event.venue]):
        problems.append("no event details found")
    if event.date is not None:
        try:
            date.fromisoformat(event.date)
        except ValueError:
            problems.append(f"date {event.date!r} is not in YYYY-MM-DD format")
    if event.fee_inr is not None and event.fee_inr < 0:
        problems.append("fee cannot be negative")
    if event.registration_url and event.registration_url not in source_text:
        problems.append("registration_url does not appear in the announcement")
    return problems


@dataclass
class Extraction:
    doc_id: str
    status: str                       # ok, invalid, refusal, max_tokens, or error
    event: EventInfo | None = None
    problems: list[str] = field(default_factory=list)
    message_id: str | None = None


async def extract_event(client: AsyncAnthropic, doc_id: str, text: str) -> Extraction:
    try:
        response = await client.messages.parse(
            model=MODEL,
            max_tokens=1024,
            system=SYSTEM,
            output_format=EventInfo,
            messages=[{"role": "user", "content": f"<announcement>\n{text}\n</announcement>"}],
        )
    except ValidationError as error:
        return Extraction(doc_id, "invalid", problems=[f"response failed validation: {error.errors()[0]['msg']}"])
    except anthropic.APIError as error:
        return Extraction(doc_id, "error", problems=[type(error).__name__])

    if response.stop_reason != "end_turn":
        return Extraction(doc_id, response.stop_reason, message_id=response.id)
    if response.parsed_output is None:
        return Extraction(doc_id, "invalid", problems=["no structured output returned"], message_id=response.id)

    problems = check_event(response.parsed_output, text)
    if problems:
        return Extraction(doc_id, "invalid", problems=problems, message_id=response.id)
    return Extraction(doc_id, "ok", event=response.parsed_output, message_id=response.id)

How It Works

1. EventInfo has only required, nullable fields, so every record has the same shape (Lessons 104 and 108). Descriptions state the formats and the missing-value behavior.
2. check_event() holds the rules a schema cannot express, including a check that the registration link really appears in the announcement. It returns a list of problems, which is empty when the event is valid.
3. Extraction is the single result type. Its status says what happened, and its problems say why, so the pipeline can count and store every outcome the same way.
4. Every failure becomes a result. A validation error, an API error, a refusal, or a truncated response each returns an Extraction rather than raising, so one bad document cannot stop the run.
5. Stop reasons are handled first. Any value other than end_turn is returned as its own status. A completed response with no parsed output is reported as invalid, so the pipeline only ever sees statuses it counts.
6. response.id records the message ID for traceability (Lessons 29 and 49).

Module 2: pipeline.py

The pipeline decides what still needs work, runs the extractions concurrently with a limit, and stores the results.

"""pipeline.py: extract event details from many announcements into SQLite."""

import asyncio
import json
import sqlite3
import time
from collections import Counter

from anthropic import AsyncAnthropic

from events_extractor import Extraction, extract_event

CONCURRENCY = 4
DB_PATH = "extractions.db"
DDL = """
CREATE TABLE IF NOT EXISTS extractions (
    doc_id       TEXT PRIMARY KEY,
    status       TEXT NOT NULL,
    data         TEXT,
    problems     TEXT NOT NULL,
    message_id   TEXT,
    extracted_at REAL NOT NULL
)
"""


def open_db(path: str) -> sqlite3.Connection:
    db = sqlite3.connect(path)
    db.execute(DDL)
    return db


def already_done(db: sqlite3.Connection, doc_id: str) -> bool:
    row = db.execute("SELECT status FROM extractions WHERE doc_id = ?", (doc_id,)).fetchone()
    return row is not None and row[0] == "ok"


def save(db: sqlite3.Connection, result: Extraction) -> None:
    data = json.dumps(result.event.model_dump(), ensure_ascii=False) if result.event else None
    db.execute(
        "INSERT OR REPLACE INTO extractions "
        "(doc_id, status, data, problems, message_id, extracted_at) VALUES (?, ?, ?, ?, ?, ?)",
        (result.doc_id, result.status, data, json.dumps(result.problems, ensure_ascii=False),
         result.message_id, time.time()),
    )


async def run(documents: dict[str, str], db_path: str = DB_PATH) -> Counter:
    db = open_db(db_path)
    todo = {doc_id: text for doc_id, text in documents.items() if not already_done(db, doc_id)}
    print(f"{len(documents) - len(todo)} already extracted, {len(todo)} to process")

    limit = asyncio.Semaphore(CONCURRENCY)
    async with AsyncAnthropic() as client:
        async def guarded(doc_id: str, text: str) -> Extraction:
            async with limit:
                return await extract_event(client, doc_id, text)

        results = await asyncio.gather(*(guarded(doc_id, text) for doc_id, text in todo.items()))

    for result in results:
        save(db, result)
    db.commit()
    db.close()
    return Counter(result.status for result in results)


DOCUMENTS = {
    "ann-001": "Free webinar: Intro to APIs on 2026-11-18 at 7 PM IST. "
               "Register at https://gocourse.example/api-webinar",
    "ann-002": "Django Weekend Workshop, 21 November 2026, Anna Hall Chennai. Fee Rs. 1,500. "
               "Sign up at https://gocourse.example/django",
    "ann-003": "Our campus fest is coming soon. Details to follow!",
    "ann-004": "SQL Bootcamp starts 2026-12-05 at the Online Studio. Fee Rs. 2,999. "
               "Register at https://gocourse.example/sql",
}

if __name__ == "__main__":
    summary = asyncio.run(run(DOCUMENTS))
    print(dict(summary))

Typical output (the exact values extracted may vary):

0 already extracted, 4 to process
{'ok': 3, 'invalid': 1}

Running the program again, without changing anything, reports the finished documents as already extracted, and retries only the failed ones:

3 already extracted, 1 to process
{'invalid': 1}

How It Works

1. open_db() creates the table if it does not exist, so the script is safe to run repeatedly.
2. already_done() makes the job idempotent: a document with status ok is skipped on later runs. Documents that failed or were refused are processed again, because a retry might succeed.
3. save() uses INSERT OR REPLACE keyed by the document ID, so each document has exactly one row, and a retry replaces the earlier failed result instead of duplicating it.
4. The semaphore limits the number of simultaneous requests to CONCURRENCY, which keeps the job within rate limits (Lesson 75).
5. The async with AsyncAnthropic() block contains all the concurrent work, so the client is closed only after every extraction has finished (Lesson 79).
6. Writes happen after gather(), in one thread, so the SQLite connection is never shared between concurrent tasks (Lesson 66).
7. ann-003 has no event details. The model returns nothing for the title, date, or venue, and the check reports "no event details found", so the document is stored as invalid for human review, not as an empty event.
8. ann-002 shows normalization by the model: "21 November 2026" should be stored as 2026-11-21, and "Rs. 1,500" as 1500. The date check in check_event() confirms the format.

Scaling Up: The Batch API

The pipeline sends one request per document. For very large volumes that do not need immediate answers, the Message Batches API (Lessons 131–140) processes requests asynchronously at a lower price. According to the documentation, structured outputs are available in batch processing with a 50% discount. The extraction function changes little: each document becomes a batch request with the same output_config format, and the results are saved when the batch completes.

Operating the Pipeline

Common Mistakes

Rerunning Everything Each Time

What happens: finished documents are paid for again, and results change between runs.

Fix: skip documents already completed, and store a status for every document.

Duplicate Rows

What happens: each retry adds a row, and counts are wrong.

Fix: key rows by a stable document ID, and replace on retry.

One Exception Stops the Run

What happens: a single API error cancels every other extraction.

Fix: return each failure as a result, and continue.

Writing to the Database from Concurrent Tasks

What happens: SQLite errors, or corrupted data from shared connections.

Fix: collect results concurrently, and write them in one place.

Storing Failures as Empty Events

What happens: missing data looks like valid records with blank fields.

Fix: store the status and problems, and keep data empty for failed records.

Best Practices

• Make runs idempotent: skip finished work, and retry failed work.
• Return a result for every document, with a clear status.
• Limit concurrency with a semaphore.
• Validate before storing, and store problems alongside the status.
• Keep one row per document, keyed by a stable ID.
• Summarize every run by status.
• Use the Batch API for very large, non-urgent volumes.

Exercise

1. Add a stored_at timestamp column and check that it updates when a document is retried.
2. Add a --retry-invalid option that reprocesses only documents with status invalid, instead of all unfinished documents.
3. Write a query that lists the most common problems across all invalid rows. Hint: the problems column stores JSON text.
4. Explain why already_done() treats only ok as finished, and what would go wrong if it treated every stored row as finished.

Step 3: load the rows in Python and count the problems:

import json
import sqlite3
from collections import Counter

db = sqlite3.connect("extractions.db")
counts: Counter = Counter()
for (problems,) in db.execute("SELECT problems FROM extractions WHERE status = 'invalid'"):
    counts.update(json.loads(problems))
for problem, count in counts.most_common(5):
    print(count, problem)

Storing problems as JSON keeps them flexible, and Python can aggregate them directly. A separate table would be the better design if you query problems frequently.

Step 4: if already_done() counted every row, a document that failed once, for example during a network outage, would never be retried, and the job would silently leave gaps in the data. Only ok rows represent finished, validated work, so every other status is retried on the next run.