Operational Context
A Python automation engineer runs the nightly rights-resolution job that decides how each digitized object publishes, and the batch stalls the moment it reaches an orphan work — an asset whose copyright holder cannot be identified, located, or contacted, yet whose term may not have lapsed. In a museum collections management system these records carry a null, blank, or placeholder rights field, so the schema-validation layer throws before the object ever reaches a routing decision, and a single unhandled record aborts an entire multi-terabyte pass. This page resolves that exact failure: it isolates orphan records inside the streamed batch, applies jurisdictional public-domain thresholds only where dated evidence exists, scores the institutional risk of publishing, and routes everything that cannot be cleared automatically to a quarantined review state — so the run finishes for the assets that do have clean rights metadata instead of dying on the first orphan. It sits inside the threshold tuning for public domain stage and feeds the same restricted-by-default contract the rest of the rights metadata and licensing automation pipeline enforces.
Root Cause Analysis
Orphan-work handling fails along three concrete vectors, each with a single root cause rather than a spread of unrelated edge cases.
First, schema-shaped brittleness. Rigid validators expect an explicit rightsStatement or license URI, so a null field, a legacy free-text string, or an institutional placeholder like "unknown" raises a KeyError or ValueError and takes the whole batch down. The record is not malformed in any interesting sense — it is simply missing rights, which is the defining property of an orphan work, and the pipeline must treat “missing” as a routable state rather than an exception.
Second, naive date heuristics. A single hard-coded cutoff year ignores that copyright term is jurisdictional: the United States clears published works at publication + 95 years, while most of the EU clears at the author’s death + 70. A record with only a publication year cannot be judged by a life-plus-70 rule, and a false “public domain” verdict on an in-copyright orphan is an unauthorized disclosure the institution cannot retract. This is the same calibration discipline worked through in the parent threshold tuning stage — the cutoff must be computed against the current year, never frozen at build time.
Third, unbounded resolution and thin provenance. Recursively chasing external authority files to identify a rightsholder exhausts memory on a large table, and even when a record passes the date test, weak provenance means the institution may be clearing a work it has no confident basis to open. The remedy on all three fronts is the same: collapse placeholders to a single orphan signal with a strict Pydantic v2 model, evaluate against a dynamic jurisdictional threshold, gate the result behind an institutional risk score, and stream the whole thing in bounded chunks.
Canonical Solution
The solution composes four gates into one constant-memory pass. Each record is normalized so that every placeholder becomes an explicit None, an explicit trusted URI short-circuits to a resolved state, a jurisdiction-aware threshold clears only records with dated evidence, and an institutional risk score can still veto a date-passing record with thin provenance. Anything that survives none of these gates terminates in quarantine — never open access. The annotated implementation runs against any iterable of raw dicts.
from __future__ import annotations
from dataclasses import dataclass
from datetime import date
from enum import Enum
from itertools import islice
from typing import Iterable, Iterator
from pydantic import BaseModel, field_validator
_PLACEHOLDERS = {"", "unknown", "n/a", "tbd", "none", "orphan"}
class RightsState(str, Enum):
RESOLVED = "resolved" # explicit, trusted rights URI already present
PUBLIC_DOMAIN = "public_domain" # cleared by jurisdictional term + risk gate
QUARANTINED = "quarantined" # orphan held for curator review — never open
@dataclass(frozen=True)
class Thresholds:
# Term lengths, not fixed years: the cutoff is computed against today's year so
# it advances automatically every January 1 as the next year's works lapse.
us_publication_term: int = 95 # US published works: publication + 95
eu_life_term: int = 70 # EU default: author's death + 70
risk_ceiling: int = 40 # max risk score still eligible for public domain
class ObjectRecord(BaseModel):
object_id: str
rights_uri: str | None = None
publication_year: int | None = None
author_death_year: int | None = None
jurisdiction: str = "US"
provenance_confidence: int = 0 # 0-100; lower means more uncertain provenance
@field_validator("rights_uri", mode="before")
@classmethod
def blank_is_orphan(cls, v: str | None) -> str | None:
# Legacy exports store placeholders where a URI belongs. Collapsing them all
# to None makes orphan detection a single `is None` test downstream instead
# of a growing set of string comparisons scattered through the pipeline.
if v is None:
return None
return None if str(v).strip().lower() in _PLACEHOLDERS else v
def _passes_public_domain(rec: ObjectRecord, cfg: Thresholds, year: int) -> bool:
# Judge each record only by the rule its evidence supports: a US record needs a
# publication year, an EU record needs a death year. A record lacking the field
# its jurisdiction requires can never clear here — it falls through to quarantine.
if rec.jurisdiction.upper() == "US" and rec.publication_year is not None:
return (year - rec.publication_year) > cfg.us_publication_term
if rec.author_death_year is not None:
return (year - rec.author_death_year) > cfg.eu_life_term
return False
def _risk_score(rec: ObjectRecord) -> int:
# Higher = riskier to publish. Thin provenance and absent dating both raise it,
# so a record can pass the date test yet still be vetoed as too risky to open.
score = 100 - rec.provenance_confidence
if rec.publication_year is None and rec.author_death_year is None:
score += 50
return score
def resolve(rec: ObjectRecord, cfg: Thresholds, today: date) -> RightsState:
if rec.rights_uri is not None:
return RightsState.RESOLVED # 1. never re-derive rights we hold
if _passes_public_domain(rec, cfg, today.year): # 2. jurisdictional term test
if _risk_score(rec) <= cfg.risk_ceiling: # 3. institutional risk veto
return RightsState.PUBLIC_DOMAIN
return RightsState.QUARANTINED # 4. terminal default, never open
def destination(state: RightsState) -> str:
# A match statement keeps routing exhaustive: adding a RightsState member without
# a branch is a visible omission rather than a silent fall-through to publication.
match state:
case RightsState.RESOLVED:
return "publish:use-existing-uri"
case RightsState.PUBLIC_DOMAIN:
return "publish:public-domain"
case RightsState.QUARANTINED:
return "queue:curator-review"
def _chunked(rows: Iterable[dict], size: int) -> Iterator[list[dict]]:
# islice pulls at most `size` items from the lazy source, so the full table is
# never materialized — only one batch is reachable at any moment.
it = iter(rows)
while batch := list(islice(it, size)):
yield batch
def resolve_batch(
rows: Iterable[dict],
cfg: Thresholds = Thresholds(),
today: date | None = None,
size: int = 5000,
) -> Iterator[tuple[str, RightsState, str]]:
today = today or date.today() # single source of truth per run
for batch in _chunked(rows, size):
for raw in batch:
rec = ObjectRecord(**raw) # placeholders already collapsed to None
state = resolve(rec, cfg, today)
yield rec.object_id, state, destination(state)Because today is captured once per run and the thresholds are term lengths rather than fixed years, the same code clears one additional cohort of works every January 1 without a redeploy. Every record reaches exactly one of three terminal states, and each state maps to a single deterministic destination, so re-running a batch is idempotent.
Those terminal states are not all permanent: a quarantined orphan can be worked back into a resolved record, and the recovery loop below traces that return path.
Edge Cases and Variants
- CSV vs. XML vs. API input.
resolve_batchis agnostic to source: feed it acsv.DictReader, aniterparsegenerator that callselem.clear()per record, or a paged API cursor as covered in polling museum APIs with Python Requests. Only the row source changes; the collapse/threshold/score loop is unchanged. - Strict vs. lenient validation. The model is strict — an unparsable
provenance_confidenceor a badobject_idraises. In a backfill, wrapObjectRecord(**raw)intry/except ValidationErrorand route the offender to the same quarantine state; in an interactive re-sync, surface the error to the operator rather than skipping the row. - Mixed or unknown jurisdiction. A record whose
jurisdictionis neitherUSnor an EU life-plus-70 territory has no rule its evidence supports, so_passes_public_domainreturnsFalseand it quarantines. Add jurisdictions by extending the term matrix onThresholds, never by loosening the default. - Date-passing but thin provenance. A work old enough to clear the term test can still carry
provenance_confidencelow enough to breachrisk_ceiling; the risk gate keeps it out of the open tier and in curator review, which is the safe direction to be wrong. - Diligent-search escalation. Quarantined orphans are not a dead end — the review queue is where a documented reasonable-search record accumulates before a curator manually assigns a RightsStatements.org orphan-works statement, after which the record re-enters as a
RESOLVEDURI. - Large tables. Feed a lazy source (a database cursor or generator), keep
sizebounded, and peak memory stays flat regardless of table size — the same reference-retention discipline as handling large CSV batches without memory leaks.
Validation
Prove the two invariants that matter — an orphan never lands in a publish state, and a date-passing record with weak provenance is vetoed rather than opened — with an assert-based test that needs no database:
# test_orphan_works.py -> run with: pytest -q test_orphan_works.py
from datetime import date
import pytest
from pydantic import ValidationError
CFG = Thresholds()
TODAY = date(2026, 7, 3)
def _row(**over):
base = {"object_id": "OBJ.1", "jurisdiction": "US", "provenance_confidence": 90}
return {**base, **over}
def test_placeholder_uri_is_treated_as_orphan():
rec = ObjectRecord(**_row(rights_uri="unknown", publication_year=2020))
assert rec.rights_uri is None # collapsed to None
assert resolve(rec, CFG, TODAY) is RightsState.QUARANTINED # in-copyright orphan
def test_old_well_provenanced_work_clears_public_domain():
rec = ObjectRecord(**_row(publication_year=1900))
assert resolve(rec, CFG, TODAY) is RightsState.PUBLIC_DOMAIN
assert destination(resolve(rec, CFG, TODAY)) == "publish:public-domain"
def test_date_passes_but_thin_provenance_is_vetoed():
rec = ObjectRecord(**_row(publication_year=1900, provenance_confidence=10))
assert resolve(rec, CFG, TODAY) is RightsState.QUARANTINED # risk gate holds it
def test_explicit_uri_short_circuits():
rec = ObjectRecord(**_row(rights_uri="https://rightsstatements.org/vocab/InC/1.0/"))
assert resolve(rec, CFG, TODAY) is RightsState.RESOLVED
def test_bad_confidence_raises_not_publishes():
with pytest.raises(ValidationError):
ObjectRecord(**_row(provenance_confidence="high"))A green run confirms that placeholders collapse to an orphan, that only an old and well-provenanced work reaches the public-domain tier, that the risk gate keeps thin-provenance records in review, and that a malformed field raises at the boundary instead of leaking into a publish decision.
Standards Alignment
The terminal states must serialize predictably into the institution’s exports. In LIDO, an orphan classification belongs in <lido:rightsWork> via <lido:rightsType>, preserving the original provenance string in <lido:creditLine> while a machine-readable statement URI is attached only once a curator resolves the record. For a quarantined orphan awaiting delivery, an IIIF Presentation API 3.0 manifest carries a restrictive RightsStatements.org URI — the orphan-works or InC-RUU statement — in its rights property, and runtime access is enforced separately rather than through the manifest. Assigning the correct URI at each state is covered in mapping RightsStatements.org to collection fields, the term arithmetic that clears a work runs through automating copyright status checks, and a time-bound release of a resolved orphan is handled by implementing embargo workflows. See the Python dataclasses documentation for the frozen immutable-configuration pattern the threshold matrix relies on.
Frequently Asked Questions
What exactly makes a record an “orphan work” in this pipeline?
An orphan work is an in-copyright (or undetermined) asset whose rightsholder cannot be identified or contacted, so it arrives with a null, blank, or placeholder rights field. The pipeline treats that missing URI as a routable state, not an exception — the record is collapsed to rights_uri = None and flows to the threshold and risk gates rather than crashing the batch.
Why quarantine an orphan instead of just publishing anything old enough?
Age alone is not clearance. A record can pass the jurisdictional term test yet carry provenance too thin to justify opening it, and a false “public domain” verdict on a work still under copyright is an unauthorized disclosure the institution cannot take back. Quarantine is the safe direction to be wrong: it delays a scan rather than exposing a rightsholder’s work.
How do the thresholds stay current without a redeploy?
Thresholds stores term lengths (publication + 95, death + 70), and resolve compares them against today.year, captured once per run. Every January 1 the same code clears one additional cohort automatically. Legal counsel adjusts a term by editing the frozen config, not the branching logic.
How does an orphan ever leave the quarantine queue?
Through documented diligent search. A curator records a reasonable-search effort, then assigns an appropriate RightsStatements.org statement — including the dedicated orphan-works or InC-RUU URIs — and the record re-enters the pipeline as a RESOLVED URI on the next pass. Nothing auto-publishes out of quarantine.
Related
- Threshold Tuning for Public Domain — parent cutoff stage
- Automating Copyright Status Checks — jurisdictional term arithmetic
- Mapping RightsStatements.org to Collection Fields — assigning the correct URI
- Implementing Embargo Workflows — time-bound release of resolved records
- Schema Validation with Pydantic — enforcing record contracts