Typed CSV: CSVW metadata and Table Schema

"""Typed CSV for the parts of CSVW and Table Schema this lesson uses: one model of a column, read from either
description, then every cell parsed by its column and every key checked, with the row and column of each error."""

import csv
import io
import re
from dataclasses import dataclass
from datetime import datetime
from decimal import Decimal


@dataclass
class Column:
    name: str
    base: str = "string"  # string, decimal, date or boolean
    required: bool = False
    pattern: str | None = None  # a regular expression the whole text must match (string)
    date_format: str | None = None  # a strptime pattern, and the form it stands for (date)
    date_form: str | None = None
    group_char: str | None = None  # removed before the number is read (decimal)
    true_false: tuple = ("true", "false")  # the texts for True and False (boolean)
    minimum: Decimal | None = None


@dataclass
class Table:
    columns: list
    primary_key: list
    foreign_keys: list  # (columns here, table there, columns there)


CSVW_DATE = {"dd": (r"\d{2}", "%d"), "MM": (r"\d{2}", "%m"), "yyyy": (r"\d{4}", "%Y")}


def from_csvw(metadata, url):
    """The table of a CSVW table group whose url is `url`."""
    table = next(t for t in metadata["tables"] if t["url"] == url)["tableSchema"]
    columns = []
    for c in table["columns"]:
        dt = c.get("datatype", "string")
        dt = {"base": dt} if isinstance(dt, str) else dt
        col = Column(c["name"], dt["base"], c.get("required", False))
        fmt = dt.get("format")
        if col.base == "string" and fmt:
            col.pattern = fmt.removeprefix("^").removesuffix("$")
        elif col.base == "date":
            parts = re.split(r"(dd|MM|yyyy)", fmt)
            col.date_format = "".join(CSVW_DATE[p][1] if p in CSVW_DATE else p for p in parts)
            col.date_form = fmt
            col.pattern = "".join(CSVW_DATE[p][0] if p in CSVW_DATE else re.escape(p) for p in parts)
        elif col.base == "decimal":
            col.group_char = (fmt or {}).get("groupChar")
        elif col.base == "boolean" and fmt:
            col.true_false = tuple(fmt.split("|"))
        if "minimum" in dt:
            col.minimum = Decimal(str(dt["minimum"]))
        columns.append(col)
    key = table.get("primaryKey", [])
    fks = [([fk["columnReference"]], fk["reference"]["resource"], [fk["reference"]["columnReference"]])
           for fk in table.get("foreignKeys", [])]
    return Table(columns, [key] if isinstance(key, str) else key, fks)


def from_table_schema(schema):
    """The table a Frictionless Table Schema describes."""
    columns = []
    for f in schema["fields"]:
        rules = f.get("constraints", {})
        base = {"number": "decimal"}.get(f.get("type", "string"), f.get("type", "string"))
        col = Column(f["name"], base, rules.get("required", False), rules.get("pattern"))
        if base == "date":
            col.date_format = f.get("format", "%Y-%m-%d")
            col.date_form = col.date_format
        col.group_char = f.get("groupChar")
        if base == "boolean":
            col.true_false = (f["trueValues"][0], f["falseValues"][0])
        if "minimum" in rules:
            col.minimum = Decimal(str(rules["minimum"]))
        columns.append(col)
    fks = [(fk["fields"], fk["reference"]["resource"], fk["reference"]["fields"]) for fk in schema.get("foreignKeys", [])]
    return Table(columns, schema.get("primaryKey", []), fks)


def parse(col, text):
    """The typed value of one cell, and the reason it is wrong (None when it is right)."""
    if text == "":
        return None, "required, but empty" if col.required else None
    if col.pattern and not re.fullmatch(col.pattern, text):
        if col.base == "date":
            return text, f"{text!r} is not a date written {col.date_form}"
        return text, f"{text!r} does not match {col.pattern}"
    if col.base == "date":
        try:
            return datetime.strptime(text, col.date_format).date(), None
        except ValueError:
            return text, f"{text!r} is not a date written {col.date_form}"
    if col.base == "boolean":
        if text not in col.true_false:
            return text, f"{text!r} is not {col.true_false[0]} or {col.true_false[1]}"
        return text == col.true_false[0], None
    if col.base == "decimal":
        plain = text.replace(col.group_char, "") if col.group_char else text
        if not re.fullmatch(r"-?\d+(?:\.\d+)?", plain):
            return text, f"{text!r} is not a decimal"
        value = Decimal(plain)
        if col.minimum is not None and value < col.minimum:
            return value, f"{text} is below the minimum {col.minimum}"
        return value, None
    return text, None


def check(text, table, others):
    """The typed rows of `text` and every error, as "row n, column: reason" (row 1 is the first after the header)."""
    reader = csv.reader(io.StringIO(text, newline=""))
    header = next(reader)
    names = [c.name for c in table.columns]
    errors = [] if header == names else [(0, 0, f"header {header} is not {names}")]
    rows = []
    for n, fields in enumerate(reader, start=1):
        row = {}
        for i, (col, cell) in enumerate(zip(table.columns, fields)):
            row[col.name], problem = parse(col, cell)
            if problem:
                errors.append((n, i, f"row {n}, {col.name}: {problem}"))
        rows.append(row)
    seen = {}
    for n, row in enumerate(rows, start=1):
        key = tuple(row[k] for k in table.primary_key)
        if key in seen:
            errors.append((n, names.index(table.primary_key[0]), f"row {n}, {'+'.join(table.primary_key)}: {key[0]!r} is already the key of row {seen[key]}"))
        seen.setdefault(key, n)
    for here, there, their in table.foreign_keys:
        keys = {tuple(r[c] for c in their) for r in others[there]}
        for n, row in enumerate(rows, start=1):
            value = tuple(row[c] for c in here)
            if None not in value and value not in keys:
                errors.append((n, names.index(here[0]), f"row {n}, {here[0]}: {value[0]!r} is not a {their[0]} of {there}"))
    return rows, [message for _, _, message in sorted(errors)]


def read_rows(text):
    """The records of a CSV text as dicts of strings, by its header."""
    return list(csv.DictReader(io.StringIO(text, newline="")))

"""invoices.csv checked against its CSVW metadata, csv-metadata.json, which also describes customers.csv."""

import json
from pathlib import Path

from typedcsv import check, from_csvw, read_rows

metadata = json.loads(Path("csv-metadata.json").read_text(encoding="utf-8"))
invoices = from_csvw(metadata, "invoices.csv")
customers = read_rows(Path("customers.csv").read_text(encoding="utf-8"))

rows, errors = check(Path("invoices.csv").read_text(encoding="utf-8"), invoices, {"customers.csv": customers})
print(f"{len(rows)} rows, {len(errors)} errors")
for error in errors:
    print(" ", error)
print("row 1 typed:", rows[0])

5 rows, 7 errors
  row 3, customer_id: 'C-07' is not a customer_id of customers.csv
  row 3, amount: 'abc' is not a decimal
  row 4, invoice_id: 'INV-003' is already the key of row 3
  row 4, paid: 'maybe' is not Y or N
  row 5, invoice_id: 'INV-05' does not match INV-[0-9]{3}
  row 5, customer_id: required, but empty
  row 5, amount: -20.00 is below the minimum 0
row 1 typed: {'invoice_id': 'INV-001', 'customer_id': 'C-01', 'issued': datetime.date(2026, 10, 4), 'amount': Decimal('1250.50'), 'paid': True, 'note': 'first order'}

{
  "$schema": "https://datapackage.org/profiles/2.0/tableschema.json",
  "fields": [
    { "name": "invoice_id", "type": "string", "constraints": { "required": true, "pattern": "INV-[0-9]{3}" } },
    { "name": "customer_id", "type": "string", "constraints": { "required": true } },
    { "name": "issued", "type": "date", "format": "%d/%m/%Y", "constraints": { "required": true } },
    { "name": "amount", "type": "number", "groupChar": ",", "constraints": { "required": true, "minimum": 0 } },
    { "name": "paid", "type": "boolean", "trueValues": ["Y"], "falseValues": ["N"] },
    { "name": "note", "type": "string" }
  ],
  "missingValues": [""],
  "primaryKey": ["invoice_id"],
  "foreignKeys": [
    { "fields": ["customer_id"], "reference": { "resource": "customers", "fields": ["customer_id"] } }
  ]
}

"""The same invoices.csv checked against a Frictionless Table Schema, invoices.schema.json."""

import json
from pathlib import Path

from typedcsv import check, from_table_schema, read_rows

schema = json.loads(Path("invoices.schema.json").read_text(encoding="utf-8"))
invoices = from_table_schema(schema)
customers = read_rows(Path("customers.csv").read_text(encoding="utf-8"))

rows, errors = check(Path("invoices.csv").read_text(encoding="utf-8"), invoices, {"customers": customers})
print(f"{len(rows)} rows, {len(errors)} errors")
for error in errors:
    print(" ", error)

5 rows, 7 errors
  row 3, customer_id: 'C-07' is not a customer_id of customers
  row 3, amount: 'abc' is not a decimal
  row 4, invoice_id: 'INV-003' is already the key of row 3
  row 4, paid: 'maybe' is not Y or N
  row 5, invoice_id: 'INV-05' does not match INV-[0-9]{3}
  row 5, customer_id: required, but empty
  row 5, amount: -20.00 is below the minimum 0

"""The rows of invoices.csv that passed every check, as JSON in the minimal mode of the W3C's csv2json mapping."""

import json
from decimal import Decimal
from pathlib import Path

from typedcsv import check, from_csvw, read_rows

metadata = json.loads(Path("csv-metadata.json").read_text(encoding="utf-8"))
customers = read_rows(Path("customers.csv").read_text(encoding="utf-8"))
rows, errors = check(Path("invoices.csv").read_text(encoding="utf-8"), from_csvw(metadata, "invoices.csv"),
                     {"customers.csv": customers})
failed = {int(e.split(",")[0].removeprefix("row ")) for e in errors}


def json_value(value):
    """The documented mapping: a decimal becomes a JSON number, a boolean true or false, a date an ISO string."""
    if isinstance(value, Decimal):
        return float(value)
    if hasattr(value, "isoformat"):
        return value.isoformat()
    return value


# A null cell gives no member at all; the rows with errors are reported, not converted.
objects = [{k: json_value(v) for k, v in row.items() if v is not None}
           for n, row in enumerate(rows, start=1) if n not in failed]
print(json.dumps(objects, indent=2, ensure_ascii=False))
print(f"converted rows {sorted(set(range(1, len(rows) + 1)) - failed)}, left out rows {sorted(failed)}")

[
  {
    "invoice_id": "INV-001",
    "customer_id": "C-01",
    "issued": "2026-10-04",
    "amount": 1250.5,
    "paid": true,
    "note": "first order"
  },
  {
    "invoice_id": "INV-002",
    "customer_id": "C-02",
    "issued": "2026-10-13",
    "amount": 99.0,
    "paid": false
  }
]
converted rows [1, 2], left out rows [3, 4, 5]

{
  "$schema": "https://json-schema.org/draft/2020-12/schema",
  "title": "One invoice, as csv2json writes it",
  "type": "object",
  "properties": {
    "invoice_id": { "type": "string", "pattern": "^INV-[0-9]{3}$" },
    "customer_id": { "type": "string" },
    "issued": { "type": "string", "format": "date" },
    "amount": { "type": "number", "minimum": 0 },
    "paid": { "type": "boolean" },
    "note": { "type": "string" }
  },
  "required": ["invoice_id", "customer_id", "issued", "amount"],
  "additionalProperties": false
}