Operational Context

A data engineer is handed a TMS (The Museum System) SQL export and told to land every object in CollectiveAccess by month-end — or to fold both systems into a single canonical record so the institution can retire one of them. The export arrives as flat rows joined across Objects, Constituents, and ThesXrefs, but CollectiveAccess does not have those tables. It has ca_objects, ca_entities, and a bundle-and-list model where a single object’s title, materials, and rights each live in their own metadata bundle, and creators are related entities rather than columns. Load the TMS rows unchanged and the importer rejects them: an accession number lands in the wrong bundle, a repeating creator collapses to one value, and a free-text rights string never becomes a machine-readable statement. This page resolves that exact task — a concrete, field-level crosswalk between the two schemas plus the Python that drives it — so the migration is a deterministic transform rather than a spreadsheet re-key. It sits inside the comparing collection schema standards topic area and applies the same schema discipline as the wider core architecture and collection taxonomy section.

Root Cause Analysis

The migration breaks for three structural reasons, not a scatter of one-off data bugs.

First, divergent entity models. TMS is a normalized relational schema: an object is one row in Objects, its makers are rows in Constituents linked through a cross-reference table, and its classification terms come from ThesXrefs. CollectiveAccess is bundle-oriented: ca_objects holds intrinsic fields (idno, preferred labels, date, credit), while creators become ca_entities joined by a relationship, and terms resolve against ca_lists. A flat TMS row therefore has to fan out across several CollectiveAccess targets, and one TMS column can legitimately map to a bundle that another column also targets — a many-to-one collision the loader must resolve deliberately rather than by last-write-wins.

Second, divergent controlled vocabularies. TMS ThesXrefs terms and Medium strings are drawn from a local thesaurus; CollectiveAccess expects list items, ideally reconciled to a shared authority. Copying the literal string preserves a term that neither system can resolve to the other’s identifier, which is why materials and object-type fields need a reconciliation step keyed on the Getty AAT rather than a straight copy — the same alignment worked through when designing museum object schemas.

Third, rights stored differently. TMS keeps rights as a RightsType label and free-text credit line; CollectiveAccess models a rights_statement bundle that wants a machine-readable URI. Moving the display label alone loses the statement’s identity. The fix on all three fronts is the same: describe every column-to-bundle move once in a data structure, fold each TMS row through it, and validate the result against a single Pydantic v2 model so a mis-mapped or blank required field fails loudly at the boundary instead of silently importing wrong.

Canonical Solution

The crosswalk is a dictionary keyed by TMS column name; each value records the fully-qualified TMS source, the CollectiveAccess bundle it feeds, the neutral canonical field both systems reconcile to, whether the bundle repeats, and which controlled vocabulary (if any) must be reconciled first. A transform function folds a flat row into a CollectiveAccess-shaped payload, and a UnifiedObject model validates the canonical record that both systems interchange through.

python
from __future__ import annotations

from dataclasses import dataclass

from pydantic import BaseModel, ConfigDict, field_validator

_KNOWN_RIGHTS_HOSTS = ("https://rightsstatements.org/", "https://creativecommons.org/")


@dataclass(frozen=True)
class FieldMap:
    # tms_column: fully-qualified source (Table.Column) in a TMS SQL export.
    # ca_bundle:  the CollectiveAccess bundle the value is imported into.
    # canonical:  field name in the neutral schema both systems reconcile to.
    # repeatable: True when the CA bundle accepts many values (a repeating bundle).
    # vocab:      controlled list whose terms must be reconciled before import.
    tms_column: str
    ca_bundle: str
    canonical: str
    repeatable: bool = False
    vocab: str | None = None


# TMS (Objects / Constituents / ThesXrefs) on the left, CollectiveAccess
# (ca_objects / ca_entities / bundles) on the right, both reconciling to a
# neutral canonical field. Keys are the raw column names in a flattened TMS row.
CROSSWALK: dict[str, FieldMap] = {
    "ObjectNumber": FieldMap(
        "Objects.ObjectNumber", "ca_objects.idno", "accession_number"),
    "Title": FieldMap(
        "Objects.Title", "ca_objects.preferred_labels.name", "title"),
    "Dated": FieldMap(
        "Objects.Dated", "ca_objects.date", "date_display"),
    "Medium": FieldMap(
        "Objects.Medium", "ca_objects.materials_technique", "materials",
        vocab="AAT"),
    "Classification": FieldMap(
        "ThesXrefs.Term", "ca_objects.type_id", "object_type", vocab="AAT"),
    "DisplayName": FieldMap(
        "Constituents.DisplayName", "ca_entities.preferred_labels.displayname",
        "creators", repeatable=True),
    "CreditLine": FieldMap(
        "Objects.CreditLine", "ca_objects.credit", "credit_line"),
    "RightsType": FieldMap(
        "Objects.RightsType", "ca_objects.rights_statement", "rights_uri",
        vocab="RightsStatements"),
}


def transform(tms_row: dict[str, object]) -> dict[str, object]:
    # Fold one flat TMS row into a CollectiveAccess-shaped payload keyed by bundle.
    # Repeatable bundles accumulate a list; single-value bundles keep the FIRST
    # non-empty source, so a many-to-one collision never silently overwrites.
    payload: dict[str, object] = {}
    for column, value in tms_row.items():
        fmap = CROSSWALK.get(column)
        if fmap is None or value in (None, ""):
            continue                                  # unmapped or empty columns are dropped
        if fmap.repeatable:
            payload.setdefault(fmap.ca_bundle, []).append(value)
        else:
            payload.setdefault(fmap.ca_bundle, value)
    return payload


# Reverse index: which canonical field each CollectiveAccess bundle feeds.
_BUNDLE_TO_CANONICAL = {fm.ca_bundle: fm.canonical for fm in CROSSWALK.values()}


class UnifiedObject(BaseModel):
    # The neutral record both TMS and CollectiveAccess reconcile to before
    # interchange. extra="forbid" turns an unmapped bundle into a loud error
    # rather than a silently dropped field.
    model_config = ConfigDict(extra="forbid", str_strip_whitespace=True)

    accession_number: str
    title: str
    date_display: str | None = None
    materials: str | None = None
    object_type: str | None = None
    creators: list[str] = []
    credit_line: str | None = None
    rights_uri: str | None = None

    @field_validator("accession_number", "title")
    @classmethod
    def required_not_blank(cls, v: str) -> str:
        if not v.strip():
            raise ValueError("required field is blank after mapping")
        return v

    @field_validator("rights_uri")
    @classmethod
    def known_rights_host(cls, v: str | None) -> str | None:
        # A migrated RightsType must resolve to a machine-readable statement URI,
        # not a free-text label carried over verbatim from the TMS column.
        if v is None:
            return None
        if not v.startswith(_KNOWN_RIGHTS_HOSTS):
            raise ValueError(f"unrecognized rights URI: {v!r}")
        return v


def to_unified(ca_payload: dict[str, object]) -> UnifiedObject:
    # Re-key the CollectiveAccess payload to canonical names and validate. Any
    # bundle without a canonical mapping keeps its bundle key and trips
    # extra="forbid", surfacing crosswalk gaps instead of dropping data.
    canonical = {_BUNDLE_TO_CANONICAL.get(bundle, bundle): value
                 for bundle, value in ca_payload.items()}
    return UnifiedObject.model_validate(canonical)

Because the crosswalk is data, adding a column is one dictionary entry rather than a new branch, and both directions of the migration — TMS into CollectiveAccess, or both into the canonical record — read the same table. The transform/to_unified split keeps the CollectiveAccess bundle shape available for the importer while the strict model guarantees no record reaches either system missing an accession number or carrying an unrecognized rights value.

TMS-to-CollectiveAccess crosswalk: three source tables reconcile through one canonical record into three bundle targets Left column: TMS Objects, Constituents and ThesXrefs tables. Centre: a Pydantic-validated canonical UnifiedObject. Right column: CollectiveAccess ca_objects, ca_entities and ca_lists/bundles. Arrows converge from the three TMS tables into the canonical record, then fan back out to the three CollectiveAccess targets, showing the many-to-one then one-to-many nature of the migration. TMS export Canonical record CollectiveAccess import crosswalk dict fan out to bundles Objects object + rights columns ObjectNumber · Title · Dated Constituents DisplayName (repeating) many creators per object ThesXrefs classification terms local thesaurus UnifiedObject canonical schema Pydantic-validated ca_objects idno · labels · date credit · rights_statement ca_entities related creators joined by relationship ca_lists / bundles controlled vocab items reconciled to AAT

Edge Cases and Variants

  • Many-to-one collisions. Both ObjectNumber and a legacy ObjectID can target ca_objects.idno. transform uses setdefault, so the first non-empty source wins and the collision is resolved by column order rather than silently overwritten. When precedence matters, order the TMS row (or the crosswalk) so the authoritative column is seen first.
  • Repeatable bundles. A single object joins to many Constituents rows. Flatten those into one TMS row as repeated DisplayName values and transform accumulates them into a creators list; a non-repeatable bundle would instead keep only the first. Never model creators as a delimited string — CollectiveAccess wants distinct related entities.
  • Vocabulary reconciliation. Medium and Classification carry vocab="AAT", a signal that the literal string must be resolved to a shared authority before import, not copied. Run those values through a reconciliation pass keyed on the Getty AAT ahead of to_unified; the crosswalk marks which fields need it so the step is never skipped by accident.
  • Rights fields. RightsType is a free-text label in TMS but must become a statement URI in ca_objects.rights_statement. The known_rights_host validator rejects anything that is not a RightsStatements.org or Creative Commons URI, forcing the label-to-URI lookup to happen rather than leaking a display string into the rights bundle.
  • CSV vs. XML vs. API export. transform is agnostic to source. Feed it a csv.DictReader row from a SQL Server export, a dict assembled from a TMS XML dump, or a page from a REST endpoint — only the row producer changes; the crosswalk and validation are identical.
  • Strict vs. lenient loads. model_config is strict, so a blank accession number or an unmapped bundle raises. In a bulk backfill wrap to_unified in try/except ValidationError and quarantine the offender; in an interactive re-sync surface the error to the operator instead of skipping the row.

Validation

Prove the two invariants that matter for a migration — the crosswalk covers every required canonical field, and a full row round-trips into a valid unified record — with an assert-based test that needs neither TMS nor CollectiveAccess running:

python
# test_crosswalk.py  ->  run with:  pytest -q test_crosswalk.py
import pytest
from pydantic import ValidationError

REQUIRED_CANONICAL = {"accession_number", "title"}


def test_crosswalk_covers_required_fields():
    covered = {fm.canonical for fm in CROSSWALK.values()}
    missing = REQUIRED_CANONICAL - covered
    assert not missing, f"crosswalk is missing required fields: {missing}"


def test_full_row_maps_and_validates():
    row = {
        "ObjectNumber": "1998.42.1",
        "Title": "Harvest at Dusk",
        "Dated": "1889",
        "Medium": "oil on canvas",
        "DisplayName": "Vincent Example",
        "RightsType": "https://rightsstatements.org/vocab/InC/1.0/",
        "Ignored": "unmapped column",           # dropped, not an error
    }
    obj = to_unified(transform(row))
    assert obj.accession_number == "1998.42.1"
    assert obj.creators == ["Vincent Example"]
    assert obj.rights_uri.endswith("/InC/1.0/")


def test_repeatable_creator_accumulates():
    row = {"ObjectNumber": "1.1", "Title": "Untitled", "DisplayName": "A. Maker"}
    payload = transform(row)
    payload["ca_entities.preferred_labels.displayname"].append("B. Maker")
    assert to_unified(payload).creators == ["A. Maker", "B. Maker"]


def test_blank_required_field_raises():
    with pytest.raises(ValidationError):
        to_unified(transform({"ObjectNumber": "   ", "Title": "x"}))


def test_free_text_rights_is_rejected():
    with pytest.raises(ValidationError):
        to_unified(transform({"ObjectNumber": "2.2", "Title": "y", "RightsType": "In Copyright"}))

A green run confirms the crosswalk is complete for required fields, that a mixed row fans out and validates, that repeating creators accumulate rather than collapse, and that a blank accession number or a free-text rights label fails at the boundary instead of importing wrong. The same discipline underpins writing custom Pydantic validators for accession numbers, where the accession field gets its own format rules.

Standards Alignment

Neither TMS nor CollectiveAccess is the interchange format — both are systems of record that should export to a shared standard. The UnifiedObject fields line up cleanly with Dublin Core: title to dc:title, creators to dc:creator, date_display to dc:date, and rights_uri to dc:rights. For richer exchange, the same record serializes to LIDO: the accession number becomes <lido:workID>, creators populate <lido:actorInRole>, materials map to <lido:termMaterialsTech> (reconciled against the Getty vocabularies), and the rights URI attaches through <lido:rightsWork>. Building the crosswalk around a canonical record rather than a point-to-point TMS-to-CollectiveAccess copy means the LIDO and Dublin Core exports fall out of the same structure — the alignment detailed in mapping LIDO to internal databases.

FAQ

Why introduce a canonical record instead of mapping TMS straight into CollectiveAccess?

A direct column-to-bundle copy hard-codes one direction and one target. A canonical UnifiedObject lets both systems reconcile to a neutral shape, which is what you want when unifying rather than replacing — and it gives you a single validated structure to serialize into LIDO or Dublin Core. The crosswalk still records the CollectiveAccess bundle, so the importer shape is never lost.

How do I handle a TMS object with several makers?

Flatten the joined Constituents rows into one TMS row carrying repeated DisplayName values. Because that column is marked repeatable, transform accumulates them into a creators list that becomes distinct ca_entities related by relationship, rather than a single delimited string CollectiveAccess cannot split back apart.

What happens to a controlled-vocabulary term that has no CollectiveAccess equivalent?

The crosswalk flags Medium and Classification with vocab="AAT", signaling that the literal must be reconciled to a shared authority before import. Run those values through a reconciliation pass first; if a term cannot be matched, quarantine the record for a curator rather than importing an unresolved string that neither system can dereference.

Does the crosswalk work for a CSV, XML, or API export?

Yes. transform consumes a plain dict, so the row producer — a csv.DictReader, a dict built from a TMS XML dump, or a REST page — is interchangeable. Only the source adapter changes; the crosswalk table and the UnifiedObject validation stay identical across all three.