From Findings to Rules: Writing the Data Contract

Module 5, Unit 1 | Lesson 2 of 3

By the end of this lesson, you will be able to:

• Map each source field from the AI usage log to a clean target field with explicit type and rule definitions (K11, S10)
• Write a Pydantic validation model that enforces normalisation rules for dates, tool names, booleans and numeric ranges (K12, S10)
• Run the validator against the full CSV and interpret the results (S9, S10)
• Export a clean, typed dataset as the first handoff artefact for your pipeline (K11, S10, B6)

In L1.1 you profiled ai_tool_usage_log.csv and documented exactly what is wrong with it: three date formats, eleven tool name variants, five boolean representations, six missing quality scores, and two negative time values. That profile is your specification. In this lesson you turn it into enforcement.

A schema is a formal description of what valid data looks like. A validator is code that checks every incoming record against that schema and either passes it through or rejects it with a clear error. Together they form a data contract — a boundary between messy source data and the clean pipeline logic that follows.

You are going to write that contract using Pydantic, the same library used by FastAPI and many production Python services.

🔑 Key term — Data contract: A formal agreement about the structure, types and constraints of data flowing between two parts of a system. A data contract makes implicit assumptions explicit and enforces them automatically, so violations produce clear errors rather than silent wrong values.

Why Pydantic?

Pydantic lets you define data models as Python classes. Each field has a type, and Pydantic enforces those types automatically. When a record violates the schema, you get a detailed ValidationError rather than a silent wrong value propagating through your pipeline.

pip install "pydantic>=2"

Curious Cat

Could you just write if statements to check each field? You could — but as the schema grows, that becomes hundreds of conditions with no structure, no standard error format, and no documentation. Pydantic gives you declarative validation (you describe what valid looks like, not how to check it), automatic type coercion, and clean error messages that tell you exactly which field failed and why. It also generates documentation automatically from the model definition.

Design the target schema

Before writing code, write the schema on paper. This is the source-to-target mapping your pipeline will enforce:

Coach Cora

Notice that time_saved_mins is allowed to be negative — you are not correcting the data, you are accepting it and flagging it for later routing. The distinction matters. Your pipeline's job at this stage is to detect and document anomalies, not to decide what the data should say. Changing a value is a business decision. That belongs in Unit 3, with a routing rule and a human approval step — not here in your validator.

Write the Pydantic model

Create a new file — schema.py:

from __future__ import annotations

from datetime import date
from enum import Enum
from typing import Optional

from pydantic import BaseModel, field_validator


class ToolName(str, Enum):
    GPT4 = "GPT-4"
    CLAUDE3 = "Claude 3"
    GEMINI_PRO = "Gemini Pro"


class FlaggedIssue(str, Enum):
    NONE = "none"
    HALLUCINATION = "hallucination"
    PII_DETECTED = "pii_detected"
    COST_OVERRUN = "cost_overrun"


TOOL_NAME_MAP = {
    "gpt-4": "GPT-4",
    "gpt4": "GPT-4",
    "chatgpt-4": "GPT-4",
    "chatgpt 4": "GPT-4",
    "gpt-4-turbo": "GPT-4",
    "claude 3": "Claude 3",
    "claude3": "Claude 3",
    "claude 3 opus": "Claude 3",
    "gemini pro": "Gemini Pro",
    "gemini-pro": "Gemini Pro",
}

DATE_FORMATS = ["%Y-%m-%d", "%d/%m/%Y", "%d-%m-%Y"]

Those three constants are the whole cleaning policy, written down in one place. ToolName and FlaggedIssue are enums rather than plain strings so that an unexpected value is an error rather than a new category appearing quietly in a report. TOOL_NAME_MAP keys are lowercased, which is why the validator lowercases before looking up — five spellings of GPT-4 collapse to one entry each.

Next, the model itself. The field declarations alone do a surprising amount: Pydantic will coerce the CSV's strings to int and float, reject anything that will not coerce, and require every field that has no default.

class ToolUsageRecord(BaseModel):
    log_id: str
    date: date
    learner_id: str
    tool_name: ToolName
    task_type: str
    model_used: str
    prompt_tokens: int
    completion_tokens: int
    response_time_ms: int
    quality_score: Optional[float] = None
    time_saved_mins: int
    flagged_issue: FlaggedIssue
    human_reviewed: bool
    cost_usd: float

Only quality_score has a default, because it is the only field allowed to be absent.

Now the validators — the rules that plain type annotations cannot express. Add these inside the class, below the fields:

    @field_validator("log_id")
    @classmethod
    def log_id_must_have_prefix(cls, v: str) -> str:
        if not v.startswith("LOG-"):
            raise ValueError(f"log_id must start with 'LOG-', got: {v!r}")
        return v

    @field_validator("date", mode="before")
    @classmethod
    def parse_date(cls, v) -> date:
        from datetime import datetime
        # Already a date (e.g. constructed in a unit test) — nothing to parse.
        if isinstance(v, date):
            return v
        for fmt in DATE_FORMATS:
            try:
                return datetime.strptime(v, fmt).date()
            except ValueError:
                continue
        raise ValueError(
            f"Unrecognised date format: {v!r}. "
            f"Expected one of: {DATE_FORMATS}"
        )

The other two validators handle the messy free-text columns you profiled in L1.1, mapping eleven spellings of three tools onto the enum and four spellings of true onto a boolean. These coerce rather than reject — the value is recoverable, so recover it.

    @field_validator("tool_name", mode="before")
    @classmethod
    def normalise_tool_name(cls, v: str) -> str:
        normalised = TOOL_NAME_MAP.get(v.strip().lower())
        if normalised is None:
            raise ValueError(
                f"Unknown tool name: {v!r}. Add it to TOOL_NAME_MAP."
            )
        return normalised

    @field_validator("human_reviewed", mode="before")
    @classmethod
    def parse_human_reviewed(cls, v) -> bool:
        if isinstance(v, bool):
            return v
        if isinstance(v, str):
            if v.strip().lower() in ("yes", "true", "1"):
                return True
            if v.strip().lower() in ("no", "false", "0"):
                return False
        raise ValueError(f"Cannot parse human_reviewed value: {v!r}")

Those three all use mode="before", because they each transform a raw string into something Pydantic can then type-check. The remaining validators run after type checking — by the time they see the value it is already an int or a float, so they only have to judge whether it is acceptable:

    @field_validator("quality_score", mode="before")
    @classmethod
    def quality_score_range(cls, v) -> Optional[float]:
        import math
        # A missing quality_score reaches us in three different shapes,
        # depending on who read the file. All three mean "no score given".
        if v is None:                                   # already None
            return None
        if v == "":                                     # csv.DictReader
            return None
        if isinstance(v, float) and math.isnan(v):      # pandas
            return None
        v_float = float(v)
        if not (1.0 <= v_float <= 5.0):
            raise ValueError(
                f"quality_score must be between 1.0 and 5.0, got: {v}"
            )
        return v_float

    @field_validator("prompt_tokens", "response_time_ms")
    @classmethod
    def must_be_positive(cls, v: int) -> int:
        if v <= 0:
            raise ValueError(f"Value must be > 0, got: {v}")
        return v

    @field_validator("completion_tokens", "cost_usd")
    @classmethod
    def must_be_non_negative(cls, v) -> float | int:
        if v < 0:
            raise ValueError(f"Value must be >= 0, got: {v}")
        return v

🔑 Key term — Field validator: A Pydantic method decorated with @field_validator that runs custom logic on a single field's value. It can transform the value (like normalising a tool name), reject it with a ValueError, or pass it through unchanged. Validators with mode="before" run before Pydantic's own type checking, which is how you handle multi-format date parsing.

Coach Cora

Look closely at the three ways quality_score can arrive empty, because this one has teeth. The validate.py you are about to write reads the CSV with pandas, and pandas turns an empty numeric cell into float('nan'). But in L1.3 you will read the same file with csv.DictReader from the standard library, and that gives you an empty string "" instead. Same file, same six cells, two completely different Python values. Handle only the pandas case and your schema passes all 30 rows today and rejects six of them the moment the pipeline reads the file a different way — pushing you over the 10% halt threshold you are about to build. A validator sits at a boundary, and boundaries are where assumptions about "how the data arrives" go to die. Handle every shape that means the same thing.

A mode="before" validator receives whatever the caller passed in — which is not always a string. parse_date starts with if isinstance(v, date): return v for exactly that reason: when you build a record by hand in a unit test (as you will in L2.1) you pass a real date object, and datetime.strptime raises TypeError on anything that is not a string. Pydantic converts a ValueError into a clean ValidationError, but it lets a TypeError propagate and crash. Making a validator accept its own output — being idempotent — is a habit worth keeping.

Run the validator against the CSV

Create validate.py and run every row through the schema:

import pandas as pd
from pydantic import ValidationError
from schema import ToolUsageRecord

df = pd.read_csv("ai_tool_usage_log.csv")

valid_records = []
invalid_records = []

for _, row in df.iterrows():
    try:
        record = ToolUsageRecord(**row.to_dict())
        valid_records.append(record)
    except ValidationError as e:
        invalid_records.append({
            "log_id": row.get("log_id", "UNKNOWN"),
            "errors": e.errors(),
        })

print(f"Valid records:   {len(valid_records)}")
print(f"Invalid records: {len(invalid_records)}")

if invalid_records:
    print("\nValidation failures:")
    for failure in invalid_records:
        print(f"\n  {failure['log_id']}:")
        for err in failure["errors"]:
            print(f"    [{err['loc'][0]}] {err['msg']}")

When you run this, every row should pass — the schema is designed to accept the data once the validators normalise it. The tool name variants get silently mapped to canonical names, dates get parsed regardless of format, and human_reviewed becomes a clean boolean everywhere.

Expected output:

Valid records:   30
Invalid records: 0

Curious Cat

If all 30 records pass, what was the point of the validator? The point is the normalisation. After validation, every record has a canonical tool_name, a proper date object, and a boolean human_reviewed. The downstream pipeline never sees the source mess. Run print(valid_records[1].tool_name) to confirm that ChatGPT-4 came out as GPT-4. That transformation is silent, automatic, and documented in the schema — not scattered across ad hoc string comparisons in your pipeline logic.

Export clean records

Add this to validate.py to write the validated, normalised data to a new file:

import json
from datetime import date

def serialise(obj):
    if isinstance(obj, date):
        return obj.isoformat()
    if hasattr(obj, "value"):   # handles Enum members
        return obj.value
    raise TypeError(f"Type {type(obj)} not serialisable")

clean_data = [r.model_dump() for r in valid_records]

with open("clean_records.json", "w") as f:
    json.dump(clean_data, f, indent=2, default=serialise)

print(f"\nClean records saved to clean_records.json ({len(clean_data)} records)")

clean_records.json is your pipeline's first formal handoff — the output of Stage 1, verified and ready for Stage 2.

Build activity

Complete both scripts so that running python validate.py produces:

1. A count of valid and invalid records
2. A printed list of any validation failures, with field name and error message for each
3. A clean_records.json file containing all 30 validated, normalised records

Then add one more validator of your own: write a @field_validator for learner_id that checks the value matches the pattern L followed by exactly three digits — for example L042. Use Python's re module. Run the validator. Do all 30 records pass?

Challenge Chase

Right now, a row that fails validation is simply logged and skipped. For a production pipeline that is not good enough. Extend validate.py to write all invalid records to an invalid_records.json quarantine file with their full error detail. Then add a halt condition: if the invalid rate exceeds 10%, raise a RuntimeError to stop the pipeline rather than continuing with degraded data. Why is an automatic halt the right behaviour at that threshold rather than a warning?

The code from this lesson

This is the schema, the validator and everything from L1.1, as the lesson leaves it. The code files are assembled from the code blocks above, so they match what you were asked to write; the dataset and a short README come with them.

Take it if you would rather read and run the code than type it out. Typing it is still the better way to learn it — this is here so a typo cannot cost you the lesson.

Checklist

• I have installed Pydantic and can import BaseModel without errors
• My schema.py defines ToolUsageRecord with all 14 fields and their types
• The parse_date validator handles all three date formats correctly
• The normalise_tool_name validator maps all 11 variants to the correct canonical name
• The parse_human_reviewed validator converts Yes, YES, No, NO and no to booleans
• Running python validate.py shows 30 valid records and 0 invalid
• clean_records.json exists and contains 30 normalised records
• In clean_records.json, tool_name values are canonical and human_reviewed is boolean
• I have written and tested the learner_id format validator

Quick Check

KSB evidence focus

• K11 — Understands relevant data governance, data privacy and security issues. The ToolUsageRecord schema is a data governance artefact. It makes every constraint on the data explicit, versioned, and enforceable. When a compliance question arises about how PII-flagged records were handled, this schema is part of the answer.

• K12 — Understands how to set up, interact with and generate APIs, databases and spreadsheets. Pydantic is the validation layer used by every major Python API framework. Understanding how to write field validators and model validators is a transferable skill that applies directly to building APIs, processing webhook payloads, and validating database inputs.

• S10 — Can carry out data processing tasks, including processing/cleaning, data transformations, and feature engineering. This lesson is applied data cleaning. The tool name normalisation, date parsing, and boolean standardisation you have written here are exactly the transformations that turn raw operational data into a format a system can trust.

Up next: Lesson 3 connects the schema to the full pipeline. You will design the stage architecture, define the data structures that flow between stages, and build a working pipeline runner that chains ingest, validate, classify, route and output — with clear slots for the agentic logic you will add in Unit 3.