Institutional Repository Strategy: Engineering FAIR Ingestion, Enrichment, and Validation Workflows #

An institutional repository earns its keep only when it stops behaving like a filing cabinet and starts behaving like a pipeline. For research data managers, academic IT teams, and Python automation engineers, the operational priority is a reproducible workflow that enforces FAIR principles at write time rather than reconstructing them at audit time. This guide sits inside the Open Science Infrastructure Planning pipeline as its deposit layer — the stage that runs after governance policy is compiled and before a dataset becomes publicly discoverable — and shows how to couple ingestion, metadata enrichment, license and retention governance, and a compliance gate into a single observable workflow. It assumes you already receive submissions from a form, an instrument, or an API and want to know exactly where each guarantee is enforced, what happens to a record that fails, and how every stage stays traceable under load. Architected this way, the repository eliminates manual curation bottlenecks while guaranteeing that an asset cannot go public until it has already passed every checkpoint.

Concept and Specification: What the Deposit Layer Must Guarantee #

The deposit pipeline binds five bodies of standard to concrete pipeline stages, and every rule the repository enforces references one of them. The DataCite Metadata Schema (version 4.6) defines the mandatory citation properties — identifier, creators, titles, publisher, publication year, resource type, rights — that a citable dataset must carry before it can mint a DOI; the field-by-field translation from internal records into that schema is the subject of Metadata Schema Mapping. Contributor identity is asserted through ORCID iDs in their full URI form, and organizational affiliation through the Research Organization Registry (ROR). Funder attribution is normalized against the Crossref Funder Registry so grant reporting rolls up to a canonical funder ID rather than a free-text agency name. Discovery metadata is serialized to the Schema.org Dataset type so harvesters and search engines can index the record, and licensing is recorded as a single SPDX License List token so a legal permission set reduces to one canonical identifier.

Those standards define shape; the repository strategy defines sequence and failure behavior. Four ordered stages carry a record from receipt to publication: staged ingestion (does the bitstream arrive intact and match a strict contract?), enrichment (can every identifier be resolved and normalized against an authoritative registry?), governance (does the license and retention window satisfy institutional and funder policy?), and the compliance gate (does the final record meet the mandatory FAIR thresholds?). A record that fails any stage is quarantined or dead-lettered with a machine-readable reason rather than dropped, so the archive can only ever contain compliant assets and every rejection is reconstructable. The structural contract at the ingestion boundary is built with Pydantic schema validation; the sections below implement one stage per concern, in the order a record must clear them.

Step-by-Step Implementation #

The pipeline advances each submission through four stages. Every stage is annotated with the compliance rationale it satisfies, and each emits a correlation ID that persists through the entire lifecycle so a single record can be traced across distributed workers.

Step 1 — Stage ingestion and validate the payload contract #

A robust ingestion layer decouples payload receipt from processing execution. A message broker (RabbitMQ, Apache Kafka, or AWS SQS) buffers incoming datasets and routes them to stateless worker pools, so a burst of grant-deadline submissions queues instead of overwhelming the pipeline. Immediately upon receipt, compute a SHA-256 checksum and persist it alongside a JSON manifest; this makes retries idempotent and turns silent bitstream corruption into a loud, catchable failure. Wrap the upload contract in a Pydantic V2 model to enforce strict schema validation at the edge, rejecting any payload missing a mandatory identifier such as dataset_id, principal_investigator, funding_agency, or data_management_plan_reference. This stage satisfies the Findable precondition — nothing enters the archive without a stable internal identifier and a verified checksum.

import hashlib
import logging
from typing import Optional

import tenacity
from pydantic import BaseModel, Field

logger = logging.getLogger("repository.ingestion")


class IngestionPayload(BaseModel):
    """Strict edge contract for an incoming deposit."""
    dataset_id: str = Field(..., pattern=r"^DS-\d{8}$")
    principal_investigator: str = Field(..., min_length=2)
    funding_agency: str
    data_management_plan_reference: Optional[str] = None
    payload_checksum: str

    @classmethod
    def validate_and_hash(cls, raw_data: dict, file_bytes: bytes) -> "IngestionPayload":
        """Verify the declared checksum against the received bytes before trusting the payload."""
        computed_hash = hashlib.sha256(file_bytes).hexdigest()
        if raw_data.get("payload_checksum") != computed_hash:
            raise ValueError("Checksum mismatch: potential bitstream corruption")
        return cls(**raw_data)


@tenacity.retry(
    stop=tenacity.stop_after_attempt(5),
    wait=tenacity.wait_exponential(multiplier=1, min=2, max=30),
    retry=tenacity.retry_if_exception_type((ConnectionError, TimeoutError)),
    reraise=True,
)
def route_to_processing_queue(manifest: IngestionPayload, correlation_id: str) -> None:
    """Publish to the broker with a correlation ID attached for end-to-end tracing."""
    logger.info(
        "Manifest %s queued for enrichment (correlation_id=%s)",
        manifest.dataset_id,
        correlation_id,
    )

Error handling here must be non-blocking and structured. Route malformed payloads to a dedicated dead-letter queue with enriched context — failure reason, validation traceback, timestamp, retry counter — and implement exponential backoff for transient storage failures while failing fast on schema violations. Academic IT teams should configure storage quotas and rate limiting at the API gateway so a bulk submission cannot saturate the pipeline; the batching mechanics for high-volume deposits are covered in async batch processing.

Step 2 — Enrich metadata against authoritative registries #

Raw institutional metadata rarely satisfies FAIR interoperability requirements on its own. The enrichment worker normalizes heterogeneous inputs into a standard schema and resolves every identifier against an authoritative registry, so that a contributor is a verified ORCID record rather than an ambiguous name string. This stage satisfies the Interoperable principle: after enrichment the record speaks DataCite Metadata Schema, its authors dereference to ORCID, and its affiliations resolve to ROR. Resolve contributor identities through the public ORCID API, merge the authoritative record with the local submission, and apply deterministic deduplication before advancing.

import logging
from typing import Any

import requests

logger = logging.getLogger("repository.enrichment")

ORCID_PUBLIC_API = "https://pub.orcid.org/v3.0"
TIMEOUT = (5, 10)  # (connect, read)


def resolve_and_enrich_orcid(name: str, orcid: str) -> dict[str, Any]:
    """Resolve a bare or URI ORCID to a verified person record with primary affiliation."""
    headers = {"Accept": "application/json"}

    person = requests.get(f"{ORCID_PUBLIC_API}/{orcid}/person", headers=headers, timeout=TIMEOUT)
    person.raise_for_status()
    person_data = person.json()

    # Employments live under a separate endpoint, not under /person.
    employments = requests.get(
        f"{ORCID_PUBLIC_API}/{orcid}/employments", headers=headers, timeout=TIMEOUT
    )
    employments.raise_for_status()
    affiliations = employments.json().get("affiliation-group", [])

    primary_affiliation = "Unknown"
    if affiliations:
        summaries = affiliations[0].get("summaries", [])
        if summaries:
            org = summaries[0].get("employment-summary", {}).get("organization", {})
            primary_affiliation = org.get("name", "Unknown")

    name_block = person_data.get("name") or {}
    given_names = name_block.get("given-names") or {}

    return {
        "display_name": given_names.get("value", name),
        "orcid": orcid if orcid.startswith("https://") else f"https://orcid.org/{orcid}",
        "primary_affiliation": primary_affiliation,
        "verified": True,
    }


def apply_crosswalk(local_metadata: dict) -> dict:
    """Merge authoritative registry data into the local record before governance."""
    enriched = local_metadata.copy()
    if enriched.get("contributor_orcid"):
        enriched["contributor"] = resolve_and_enrich_orcid(
            enriched.get("contributor_name", ""),
            enriched["contributor_orcid"],
        )
    return enriched

For grant-funded datasets, cross-reference the funding metadata against the Crossref Funder Registry so reporting structures match the compliance baselines described in Funder Mandate Alignment across multi-institutional consortia. The resolver control plane — circuit breakers, secondary registries, and cache fallback when a registry is slow or down — is detailed in API routing and fallbacks.

Step 3 — Assign licenses and enforce retention governance #

Automated license assignment and retention scheduling are the backbone of institutional data governance, and neither is encoded by any metadata standard for you. Rather than hard-coding a license string, evaluate the dataset against an institutional policy matrix and resolve it to an SPDX License List identifier mapped to a machine-readable license.json payload. Retention must be enforced programmatically: a dataset tagged with a grant identifier or regulatory classification drives lifecycle state transitions (ACTIVE → ARCHIVAL → DECOMMISSIONED or COLD_STORAGE), evaluated by a cron-driven worker that reads retention_expiry, writes an audit log, and routes expired artifacts to secure deletion or perpetual cold storage. This stage satisfies the Reusable principle — a dataset without a clear, machine-readable license and a defined retention window is not reusable.

from datetime import datetime, timedelta, timezone
from enum import Enum


class RetentionState(str, Enum):
    ACTIVE = "active"
    ARCHIVAL = "archival"
    DECOMMISSIONED = "decommissioned"
    COLD_STORAGE = "cold_storage"


def evaluate_retention_policy(dataset: dict, policy_matrix: dict[str, int]) -> dict:
    """Assign an SPDX license and compute the retention expiry from institutional policy."""
    grant_type = dataset.get("funding_agency", "DEFAULT")
    retention_years = policy_matrix.get(grant_type, 10)

    created_at = datetime.fromisoformat(dataset["created_at"])
    if created_at.tzinfo is None:
        created_at = created_at.replace(tzinfo=timezone.utc)
    expiry_date = created_at + timedelta(days=retention_years * 365)

    dataset.update(
        {
            "retention_state": RetentionState.ACTIVE.value,
            "retention_expiry": expiry_date.isoformat(),
            "license_spdx": dataset.get("license", "CC-BY-4.0"),
        }
    )
    return dataset

The selection and encoding of the license token itself — the mapping from institutional policy to a canonical SPDX identifier and its resolvable URI — is covered in Open License Configuration. When a dataset is domain-specific or sensitive, routing logic should direct it to a certified disciplinary repository while keeping institutional metadata in sync; the decision criteria for that split are worked through in choosing the right repository for grant-funded projects.

Step 4 — Run the FAIR compliance gate and emit telemetry #

The final stage is an automated gate that no record crosses without clearing schema checks, identifier resolution tests, and license compatibility scans. Emit structured telemetry with OpenTelemetry to capture pipeline latency, validation failure rates, and enrichment success ratios, and propagate the correlation ID from Step 1 across every span so a failing record can be traced end to end. A record that fails a mandatory FAIR threshold — missing persistent identifier, unresolved contributor, or incompatible license — is rejected to a remediation ticket routed back to the originating team, while a passing record is serialized to the repository API. This is the checkpoint that operationalizes the policy compiled in Data Governance Frameworks.

import logging

from jsonschema import ValidationError, validate

logger = logging.getLogger("repository.compliance_gate")

DATACITE_MINIMAL_SCHEMA = {
    "type": "object",
    "properties": {
        "identifier": {"type": "string", "pattern": r"^10\.\d{4,9}/[-._;()/:a-zA-Z0-9]+$"},
        "creators": {"type": "array", "minItems": 1},
        "titles": {"type": "array", "minItems": 1},
        "publisher": {"type": "string"},
        "publicationYear": {"type": "integer"},
        "rightsList": {"type": "array", "minItems": 1},
    },
    "required": ["identifier", "creators", "titles", "publisher", "publicationYear", "rightsList"],
}


def validate_fair_compliance(record: dict) -> bool:
    """Assert the record meets the mandatory FAIR thresholds before publication."""
    try:
        validate(instance=record, schema=DATACITE_MINIMAL_SCHEMA)
        # Business logic beyond structural validation: rights must be a resolvable HTTPS URI.
        if not record["rightsList"][0].get("rightsURI", "").startswith("https://"):
            raise ValueError("rightsURI must be a resolvable HTTPS endpoint")
        logger.info("FAIR gate PASSED: record cleared for repository publication.")
        return True
    except (ValidationError, ValueError) as exc:
        logger.error("FAIR gate FAILED: %s", exc)
        return False

Pipeline Stage Reference Matrix #

Every stage binds a governing standard to an enforcement point, a decision rule, and the action taken on failure. Use this matrix to reconcile a rejected record against the exact stage that stopped it.

Error Handling and Edge Cases #

A deposit pipeline is judged on how it fails, not only on how it passes. Each stage separates transient failures, which are retried, from permanent failures, which are quarantined for human remediation, and never conflates the two.

• Transient registry failures (HTTP 429/5xx during ORCID or DataCite resolution in Step 2) are retried with jittered exponential backoff honoring the Retry-After header. When retries are exhausted, route the record to a dead-letter queue keyed by dataset_id with a resolution_pending status rather than aborting the batch.
• Checksum mismatches (Step 1) are permanent and security-relevant. Quarantine the payload with the declared and computed hashes side by side; a mismatch is either corruption in transit or a tampered upload, and neither should ever advance.
• Malformed records (ValidationError at Step 1) are permanent. Serialize the Pydantic error alongside the raw payload into a quarantine store so a curator sees exactly which field failed which constraint. Never silently coerce.
• Policy rejections (Step 3 license outside the allowlist) are business-rule failures, not exceptions. Record the rejecting rule on the audit ledger and notify the depositor; these need a policy decision, not a retry.
• Partial batches must be transactional at the record level. One quarantined submission must never abort its siblings in the same run — accumulate per-record outcomes and emit a summary at the end so a single bad record does not poison a grant-deadline upload.

Verification and Testing #

Assert deposit behavior the same way you assert application logic — with tests that exercise both the pass path and every rejection path. The following covers the Step 1 checksum guard and the Step 4 compliance gate, the two stages most likely to let a bad record through if they regress.

import hashlib

import pytest
from pydantic import ValidationError as PydanticError


def test_checksum_guard_rejects_corruption() -> None:
    file_bytes = b"instrument-export-v1"
    good_hash = hashlib.sha256(file_bytes).hexdigest()

    # Valid: declared checksum matches the received bytes.
    payload = IngestionPayload.validate_and_hash(
        {
            "dataset_id": "DS-20260701",
            "principal_investigator": "Ada Lovelace",
            "funding_agency": "NIH",
            "payload_checksum": good_hash,
        },
        file_bytes,
    )
    assert payload.dataset_id == "DS-20260701"

    # Rejected: bitstream corruption surfaces as a ValueError, never a silent pass.
    with pytest.raises(ValueError, match="Checksum mismatch"):
        IngestionPayload.validate_and_hash(
            {"dataset_id": "DS-20260701", "principal_investigator": "Ada Lovelace",
             "funding_agency": "NIH", "payload_checksum": "deadbeef"},
            file_bytes,
        )

    # Rejected: malformed dataset_id fails the strict contract at the edge.
    with pytest.raises(PydanticError):
        IngestionPayload.validate_and_hash(
            {"dataset_id": "bad-id", "principal_investigator": "Ada Lovelace",
             "funding_agency": "NIH", "payload_checksum": good_hash},
            file_bytes,
        )


def test_fair_gate_blocks_unresolvable_rights() -> None:
    base = {
        "identifier": "10.5281/zenodo.7654321",
        "creators": [{"name": "Lovelace, Ada"}],
        "titles": [{"title": "Analytical Engine Notes"}],
        "publisher": "Institutional Repository",
        "publicationYear": 2026,
        "rightsList": [{"rights": "CC-BY-4.0", "rightsURI": "https://creativecommons.org/licenses/by/4.0/"}],
    }
    assert validate_fair_compliance(base) is True

    # Rejected: a non-HTTPS rights URI cannot dereference and fails the gate.
    broken = {**base, "rightsList": [{"rights": "CC-BY-4.0", "rightsURI": "not-a-url"}]}
    assert validate_fair_compliance(broken) is False

A passing run emits a contiguous ladder of stage logs; the presence of every expected line is itself a CI-grepable assertion:

2026-07-02 09:14:01 [INFO] repository.ingestion       | Manifest DS-20260701 queued for enrichment (correlation_id=7c1f…)
2026-07-02 09:14:02 [INFO] repository.enrichment       | ORCID resolved and normalized to URI form
2026-07-02 09:14:02 [INFO] repository.compliance_gate  | FAIR gate PASSED: record cleared for repository publication.

Gotchas and Known Pitfalls #

• Bare ORCID vs. full URI. A record carrying 0000-0002-1825-0097 is not interchangeable with https://orcid.org/0000-0002-1825-0097. The enrichment step normalizes to the URI form, but if you skip that normalization the Dataset document you emit will have an author.identifier that does not dereference.
• Naive vs. timezone-aware datetimes. Parsing created_at with datetime.fromisoformat yields a naive datetime when the source string has no offset, and adding a timedelta to it produces a naive retention_expiry that raises TypeError the moment the retention worker compares it against a timezone-aware now. Normalize to UTC at ingestion, as Step 3 does explicitly.
• Retrying non-idempotent deposits. The broker makes ingestion retries safe only because the checksum manifest de-duplicates re-delivered messages. If your repository API does not de-duplicate on dataset_id, a retried publish after a server-side commit double-registers the DOI; confirm idempotency before enabling retries on the publish call.
• Checksum computed after transformation. Hash the bytes exactly as received, before any decompression or re-encoding. Computing the SHA-256 over a re-serialized payload defeats the integrity guarantee — you will validate the checksum against your own transformation, not the submitter’s original bitstream.
• License allowlist drift. The SPDX identifier set that Step 3 assigns and the set that Step 4 accepts must be a single shared constant. When institutional policy adds a license and only one side is updated, valid records are quarantined at the gate for carrying a token the assignment step just wrote.

Frequently Asked Questions #

Where does the repository deposit layer sit relative to the rest of the pipeline? #

It is the stage that runs after governance policy is compiled and before public discovery — the deposit layer of the Open Science Infrastructure Planning pipeline. Ingestion and enrichment run first, license and retention governance third, and only a record that clears the FAIR compliance gate is serialized to the repository API and made discoverable.

Should sensitive or domain-specific datasets go to the institutional repository or a disciplinary one? #

Route them to a certified disciplinary repository when the domain expects it — genomics to a sequence archive, crystallography to a structure database — while keeping institutional metadata synchronized so the record stays discoverable from both. The Step 3 routing logic makes that decision from the dataset’s classification tags; the full decision matrix is in choosing the right repository for grant-funded projects.

How does the pipeline stay traceable when workers are distributed? #

Every submission is assigned a correlation ID at ingestion that is attached to the broker message headers and propagated as an OpenTelemetry span attribute through enrichment, governance, and the compliance gate. A single failing record can then be reconstructed end to end from its ID, and stage-level metrics — latency, failure rate, enrichment success ratio — roll up per pipeline without losing the per-record thread.

What is the difference between a quarantined record and a dead-lettered one? #

A quarantined record failed a permanent rule — a checksum mismatch, a malformed identifier, a disallowed license — and needs a human to correct the payload. A dead-lettered record failed a transient dependency, typically ORCID or DataCite resolution, and is expected to succeed on replay once the registry recovers. Keeping the two queues separate stops a registry outage from generating remediation tickets no curator can action.

Related Guides #

• Open Science Infrastructure Planning — the parent overview showing how this deposit layer composes with governance, funder, and license stages.
• Data Governance Frameworks — the policy layer that compiles the rules this pipeline enforces at deposit time.
• Funder Mandate Alignment — the compliance baselines the Step 2 funder crosswalk reconciles against.
• Open License Configuration — how the SPDX license tokens assigned in Step 3 are selected and encoded.
• Choosing the right repository for grant-funded projects — the routing decision behind sending a dataset to an institutional versus a disciplinary archive.