CSV to Database Sync Strategies for Museum Collections

Within the broader automated record ingestion and sync workflows pipeline, this stage is the flat-file adapter: it turns a tabular export — a donor spreadsheet, a legacy Text Management System (TMS) dump, or a nightly CollectiveAccess extract — into validated, deduplicated, standards-conformant rows inside the collection management system. Where the async ingestion pipeline owns born-structured API payloads, this adapter owns the messy reality of CSV: inconsistent delimiters, mixed encodings, and free-text rights values that must be normalized before a single row is committed. This page specifies the encoding-detection front door, the streaming reader that keeps memory flat, the Pydantic contract that gates every row, and the idempotent upsert that makes re-running a failed sync safe.

The problem CSV ingestion solves is narrow but unforgiving. A naive pandas.read_csv() loads the entire export into memory, chokes on a stray utf-16 byte-order mark, and silently coerces an accession number like 007.1994 into a float. A production sync instead streams rows in bounded chunks, validates each against a strict schema, routes malformed rows to a durable dead-letter queue, and upserts the survivors so a partial failure never leaves half-written records behind.

Workflow Context

Museum collection exports rarely arrive database-ready. Legacy CMS platforms emit CSV with semicolon or tab delimiters depending on the operator’s locale, encode diacritics in latin-1 or utf-8-sig, and stuff rights information into free-text columns that read "in copyright" in one export and "© rights reserved" in the next. Treating ingestion as a single blocking INSERT loop couples every one of these defects to the primary write path: one malformed row aborts the transaction and the whole nightly sync fails silently.

The design in this guide treats CSV ingestion as a stateful pipeline with four independently bounded stages — detect, stream, validate, commit — each of which can reject a row without halting the ones behind it. Detection normalizes encoding and delimiter before parsing. Streaming yields fixed-size chunks so memory stays flat whether the export holds two hundred rows or two hundred thousand. Validation enforces the metadata contract and splits the stream. The commit stage performs an idempotent bulk upsert keyed on accession number, so retrying an interrupted run never duplicates a record. Free-text materials and place names that survive validation still need identifier resolution — that is the job of implementing Getty AAT & TGN downstream.

Pipeline Architecture & Data Flow

The reader holds only one chunk of rows in memory at a time, so restarting a failed run costs nothing but re-reading the file from a checkpointed offset. Validation runs off the write path, and the dead-letter queue preserves every rejected row with its structured error list so nothing is silently dropped. Each arrow below is an explicit gate, not a best-effort handoff.

The streaming gate bounds memory; the validation gate enforces the metadata contract; the upsert gate enforces idempotency at the database boundary. This mirrors the async pipeline’s topology so both adapters converge on the same commit and dead-letter guarantees.

Prerequisites

Before wiring the sync, confirm the following are in place:

Python 3.9+ for the PEP 604 | union syntax and asyncio features used throughout.
asyncpg (0.29+) for high-throughput PostgreSQL bulk upserts with server-side prepared statements.
aiofiles (23+) for non-blocking disk I/O, and aiocsv (2+) to parse rows directly off the async file handle — the stdlib csv reader is synchronous and cannot be driven with async for.
pydantic (v2) for the row contract; this page uses the v2 API (model_config, field_validator, model_dump) — see the pydantic v2 documentation for migration notes.
charset-normalizer (3+) to detect encoding when the export’s declared charset is absent or wrong.
Target PostgreSQL write access with a unique constraint on object_id so ON CONFLICT can drive deterministic conflict resolution.
Standards endpoints for rights normalization: legacy free-text values map to RightsStatements.org URIs, surfaced downstream through the IIIF rights property defined in the IIIF Presentation API 3.0 specification; the LIDO schema is the interchange envelope the validated rows conform to.
A dead-letter store — a quarantine table or JSON log that accepts the raw row, the error list, and a UTC timestamp.

Schema & Field Reference

The row contract is the single source of truth for what a valid record looks like. Raw CSV fields rarely match institutional standards, so a strict Pydantic v2 model coerces types, enforces presence, and rejects unknown columns before any database work. Because every CSV cell arrives as a string, the model does explicit, auditable coercion rather than trusting a dataframe’s silent type inference.

Field	Type	Constraint	Purpose
`object_id`	`str`	required, unique	Idempotency key for the upsert path
`title`	`str`	required, non-empty	Primary descriptive field
`creator`	`str \| None`	optional	Attribution, nullable for unknown makers
`rights_status`	`str`	required, RightsStatements.org URI	Machine-readable rights assertion
`date_modified`	`date`	required, ISO 8601	Drives conflict resolution and audit trail

python

from datetime import date
from pydantic import BaseModel, ConfigDict, field_validator

RIGHTS_URI_PREFIX = "https://rightsstatements.org/vocab/"

class CollectionRow(BaseModel):
    """Strict contract for one CSV row destined for the CMS."""
    model_config = ConfigDict(str_strip_whitespace=True, extra="forbid")

    object_id: str
    title: str
    creator: str | None = None
    rights_status: str
    date_modified: date

    @field_validator("object_id", "title")
    @classmethod
    def not_blank(cls, value: str) -> str:
        # Reject rows where a required column is present but empty —
        # a blank accession number would corrupt the idempotency key.
        if not value:
            raise ValueError("required field is blank")
        return value

    @field_validator("rights_status")
    @classmethod
    def must_be_rights_uri(cls, value: str) -> str:
        # Enforce a resolvable RightsStatements.org URI, never free text.
        if not value.startswith(RIGHTS_URI_PREFIX):
            raise ValueError(f"rights_status must be a {RIGHTS_URI_PREFIX} URI")
        return value

With extra="forbid", an unmapped extra column in the export is rejected rather than silently ignored, so schema drift in the source surfaces at the validation gate instead of writing partial records.

Step-by-Step Implementation

1. Detect encoding and delimiter before parsing

A single mis-detected encoding turns an entire export into replacement characters. Sniff the charset from the raw bytes and the delimiter from a header sample before opening the file for streaming.

python

import csv
from charset_normalizer import from_path

def sniff_dialect(filepath: str) -> tuple[str, str]:
    """Return (encoding, delimiter) for a CSV export of unknown origin."""
    best = from_path(filepath).best()
    encoding = best.encoding if best else "utf-8-sig"
    # Read a small header sample in the detected encoding to sniff the delimiter.
    with open(filepath, encoding=encoding, newline="") as fh:
        sample = fh.read(8192)
    dialect = csv.Sniffer().sniff(sample, delimiters=",;\t|")
    return encoding, dialect.delimiter

Edge cases. TMS exports frequently ship as utf-16 with a BOM; charset-normalizer catches these where a hard-coded utf-8 open would raise UnicodeDecodeError. If Sniffer raises csv.Error on a single-column file, fall back to a comma. For XML sources (LIDO or EAD) the delimiter question is moot — parse with lxml.etree.iterparse instead — and for API payloads the transport is already structured JSON, so skip straight to validation.

2. Stream rows in memory-safe chunks

asyncpg and aiofiles keep the event loop responsive while the sync runs. aiocsv.AsyncDictReader parses rows directly off the async handle, and the generator yields fixed-size batches so peak memory is a function of chunk size, not file size.

python

import aiofiles
from aiocsv import AsyncDictReader
from collections.abc import AsyncGenerator

BATCH_SIZE: int = 500

async def stream_csv_chunks(
    filepath: str, encoding: str, delimiter: str, chunk_size: int
) -> AsyncGenerator[list[dict[str, str]], None]:
    """Yield CSV rows in memory-safe chunks using async file I/O."""
    async with aiofiles.open(filepath, mode="r", encoding=encoding, newline="") as fh:
        batch: list[dict[str, str]] = []
        async for row in AsyncDictReader(fh, delimiter=delimiter):
            batch.append(dict(row))
            if len(batch) >= chunk_size:
                yield batch
                # Rebind to a fresh list; clearing would mutate the chunk
                # already handed to the consumer.
                batch = []
        if batch:
            yield batch

Edge cases. Generator lifecycle is where large syncs leak — accumulating validated rows in an outer list defeats the whole design. The deep dive on handling large CSV batches without memory leaks covers generator scope, the batch = [] rebind rationale, and profiling flat memory under load.

3. Validate each chunk against the contract

Validation splits the stream. Conforming rows become CollectionRow instances; failures carry their structured error list to the dead-letter routine in step 5 without aborting the chunk.

python

from pydantic import ValidationError

def validate_chunk(
    rows: list[dict[str, str]]
) -> tuple[list[CollectionRow], list[dict[str, object]]]:
    """Split a raw chunk into (valid records, quarantined rows)."""
    valid: list[CollectionRow] = []
    rejected: list[dict[str, object]] = []
    for row in rows:
        try:
            valid.append(CollectionRow.model_validate(row))
        except ValidationError as exc:
            # Preserve the offending row plus field-level reasons for curators.
            rejected.append({"row": row, "errors": exc.errors(include_url=False)})
    return valid, rejected

Edge cases. For a lenient first-pass migration, relax the model with str | None defaults and a nightly reconciliation report rather than hard rejection; for the authoritative production sync keep extra="forbid" strict so no unmapped column is ever written.

4. Resolve legacy rights values to RightsStatements.org URIs

Free-text rights columns must become resolvable URIs before validation accepts them. A deterministic classifier maps known legacy phrases; anything unrecognized is left for a curator rather than guessed.

python

LEGACY_RIGHTS = {
    "in copyright": "https://rightsstatements.org/vocab/InC/1.0/",
    "© rights reserved": "https://rightsstatements.org/vocab/InC/1.0/",
    "public domain": "https://rightsstatements.org/vocab/NoC-US/1.0/",
    "no known copyright": "https://rightsstatements.org/vocab/NKC/1.0/",
}

def normalize_rights(raw: str) -> str | None:
    """Map a legacy free-text rights value to a RightsStatements.org URI."""
    return LEGACY_RIGHTS.get(raw.strip().casefold())

Apply normalize_rights to the raw row before model_validate, so a mapped URI passes the must_be_rights_uri check and an unmapped value is quarantined for review. See rights and access routing below for how the resolved URI drives visibility.

5. Upsert survivors and route failures

The commit stage runs each chunk inside a single transaction. ON CONFLICT (object_id) makes the write idempotent: a retried batch updates in place instead of raising a duplicate-key error.

python

import asyncpg

async def upsert_batch(pool: asyncpg.Pool, records: list[CollectionRow]) -> None:
    """Execute atomic batch upserts with conflict resolution."""
    query = """
        INSERT INTO collection_records (object_id, title, creator, rights_status, date_modified)
        VALUES ($1, $2, $3, $4, $5)
        ON CONFLICT (object_id) DO UPDATE SET
            title = EXCLUDED.title,
            creator = EXCLUDED.creator,
            rights_status = EXCLUDED.rights_status,
            date_modified = EXCLUDED.date_modified
    """
    async with pool.acquire() as conn:
        async with conn.transaction():
            await conn.executemany(
                query,
                [
                    (r.object_id, r.title, r.creator, r.rights_status, r.date_modified)
                    for r in records
                ],
            )

async def quarantine(pool: asyncpg.Pool, rejected: list[dict[str, object]]) -> None:
    """Persist rejected rows durably so nothing is silently dropped."""
    import json
    from datetime import datetime, timezone
    async with pool.acquire() as conn:
        await conn.executemany(
            "INSERT INTO quarantine (payload, errors, quarantined_at) VALUES ($1, $2, $3)",
            [
                (json.dumps(item["row"]), json.dumps(item["errors"]),
                 datetime.now(timezone.utc))
                for item in rejected
            ],
        )

Edge cases. Prefer executemany with prepared statements over string-built multi-row INSERTs — it reuses the server-side plan and sidesteps injection. When a connection times out mid-chunk, retry the whole chunk with exponential backoff and jitter; because the upsert is idempotent, replaying an already-committed chunk is harmless. Malformed rows serialize to the durable quarantine table so they survive a service restart and surface in the curator’s daily reconciliation report.

Rights and Access Routing

Rights metadata decides visibility, not just descriptive completeness. Once normalize_rights resolves a legacy value to a RightsStatements.org URI, that URI determines the access tier the record receives at commit time. A record carrying InC/1.0/ (in copyright) commits at a metadata-only tier — the descriptive record is public but the media asset is gated — while NoC-US/1.0/ and NKC/1.0/ records may expose full-resolution media. The resolved URI is later written into the IIIF manifest rights property so every downstream viewer enforces the same policy the sync asserted.

Unmapped rights values must never default to “public.” Route them to the quarantine queue exactly as a schema failure would be routed, and let a curator assign the correct URI before the record becomes resolvable. Embargo handling — records that should stay dark until a future release date — is a separate routing dimension; the pattern for that lives in implementing embargo workflows, and Creative Commons value mapping is covered in routing Creative Commons licenses.

Verification and Testing

Before pointing the sync at a production database, prove the four gates with assert-based tests and a dry run. The test below exercises a valid row, a blank-key rejection, and a rights-URI rejection in isolation.

python

import asyncio

def test_contract() -> None:
    # 1. A well-formed row validates.
    good = CollectionRow.model_validate({
        "object_id": "007.1994",
        "title": "Portrait of a Collector",
        "creator": "Unknown",
        "rights_status": "https://rightsstatements.org/vocab/InC/1.0/",
        "date_modified": "2026-06-01",
    })
    assert good.object_id == "007.1994"

    # 2. A blank accession number is rejected, not coerced.
    try:
        CollectionRow.model_validate({
            "object_id": "", "title": "x",
            "rights_status": "https://rightsstatements.org/vocab/InC/1.0/",
            "date_modified": "2026-06-01",
        })
        assert False, "blank object_id should have raised"
    except ValidationError:
        pass

    # 3. Free-text rights is rejected before any DB write.
    assert normalize_rights("In Copyright") == "https://rightsstatements.org/vocab/InC/1.0/"
    assert normalize_rights("ask the registrar") is None

if __name__ == "__main__":
    test_contract()
    print("contract OK")

Run a full dry run that streams and validates the real export but skips the commit, so you can count survivors and quarantined rows before writing anything:

bash

python -m sync.dryrun --file exports/nightly.csv --chunk-size 500 --no-commit

A clean run reports zero unexpected rejections and a survivor count matching the export’s row total minus known-bad rows. Any surprise in that delta is schema drift to investigate before the production sync.

Where to Go Deeper

Handling Large CSV Batches Without Memory Leaks — diagnoses the generator-lifecycle and accumulation bugs that make a streaming sync’s memory scale with file size, and shows how to keep the profile flat under a multi-hundred-thousand-row campaign.

FAQ

Why do accession numbers like `007.1994` get mangled on import?

Because a dataframe-style reader infers 007.1994 as a float and drops the leading zero, corrupting the idempotency key. Streaming every cell as a string through aiocsv and coercing explicitly in the Pydantic model — never letting a numeric inference touch the identifier — preserves the exact source value.

My whole nightly sync fails when one row is malformed. How do I isolate it?

You are committing inside one giant transaction. Validate per chunk with validate_chunk, route each ValidationError to the durable quarantine table, and upsert only the survivors. One bad row then costs one quarantined row and a line in the reconciliation report, not the entire run.

The export opens fine in a spreadsheet but crashes my reader with `UnicodeDecodeError`.

The declared or assumed encoding is wrong — TMS exports are frequently utf-16 with a BOM or latin-1 with diacritics. Detect the charset from the raw bytes with charset-normalizer before opening the file, as in sniff_dialect, rather than hard-coding utf-8.

Is re-running an interrupted sync safe, or will it create duplicates?

Safe, provided the write uses ON CONFLICT (object_id) DO UPDATE. The upsert is idempotent: a row committed on the first attempt is updated in place on the retry, so replaying a partially completed run never duplicates records. This is what lets you retry a timed-out chunk without reconciliation.

How should I handle rights values I have never seen before?

Never default them to public. normalize_rights returns None for unrecognized values, which fails validation and quarantines the row exactly like a schema error. A curator assigns the correct RightsStatements.org URI before the record becomes resolvable.

Automated Record Ingestion & Sync Workflows — parent pipeline overview
Handling Large CSV Batches Without Memory Leaks — flat-memory streaming deep dive
Building Async Ingestion Pipelines — born-structured API variant
Schema Validation with Pydantic — strict v2 contract design
Automating OCR Metadata Extraction — enriching scanned source fields

CSV to Database Sync Strategies for Museum Collections

Workflow Context #

Pipeline Architecture & Data Flow #

Prerequisites #

Schema & Field Reference #

Step-by-Step Implementation #

1. Detect encoding and delimiter before parsing #

2. Stream rows in memory-safe chunks #

3. Validate each chunk against the contract #

4. Resolve legacy rights values to RightsStatements.org URIs #

5. Upsert survivors and route failures #

Rights and Access Routing #

Verification and Testing #

Where to Go Deeper #

FAQ #

Why do accession numbers like 007.1994 get mangled on import? #

My whole nightly sync fails when one row is malformed. How do I isolate it? #

The export opens fine in a spreadsheet but crashes my reader with UnicodeDecodeError. #

Is re-running an interrupted sync safe, or will it create duplicates? #

How should I handle rights values I have never seen before? #

Related #

Explore this section

Workflow Context

Pipeline Architecture & Data Flow

Prerequisites

Schema & Field Reference

Step-by-Step Implementation

1. Detect encoding and delimiter before parsing

2. Stream rows in memory-safe chunks

3. Validate each chunk against the contract

4. Resolve legacy rights values to RightsStatements.org URIs

5. Upsert survivors and route failures

Rights and Access Routing

Verification and Testing

Where to Go Deeper

FAQ

Why do accession numbers like `007.1994` get mangled on import?

My whole nightly sync fails when one row is malformed. How do I isolate it?

The export opens fine in a spreadsheet but crashes my reader with `UnicodeDecodeError`.

Is re-running an interrupted sync safe, or will it create duplicates?

How should I handle rights values I have never seen before?

Related