FAIR Data Automation Research Data Management

Building a FAIR Deposit Pipeline in Prefect

This is the concrete build behind choosing Prefect to orchestrate FAIR deposits: a flow that takes a batch of submissions and drives each one through parse, validate, mint DOI, deposit, and reconcile, with per-task retries, concurrency limits that respect upstream API ceilings, and failure isolation so one bad dataset never aborts the batch. It assumes you have already weighed the orchestrators in Pipeline Orchestration for FAIR Deposit Workflows, that you have Prefect installed (pip install prefect), and that you are comfortable with Python 3.10+ and the Pydantic V2 API. The goal is a flow whose correctness does not depend on the happy path: transient registry and repository errors are retried with backoff, permanent failures are quarantined, and a retried task can never double-mint or double-deposit.

The pipeline is deliberately built so the business logic lives in ordinary functions and Prefect adds only scheduling, retries, and observability on top. Each @task is independently testable, and the @flow is the thin layer that wires them together and isolates failures per dataset.

Prefect deposit flow as a directed acyclic graph Five Prefect tasks run in order: parse, validate, mint DOI with retries, deposit with retries, and reconcile. The mint and deposit tasks carry retry policies, and a failing deposit branches down into a quarantine store instead of advancing to reconcile. parse no retry validate Pydantic V2 mint DOI retries = 5 deposit retries = 5 reconcile retries = 3 final fail quarantine dead-letter store

Task Retry and Failure Policy

Each task earns a retry policy from the kind of failure it can hit. Parse and validate fail on malformed or non-conformant input — a permanent, record-level fault that must be quarantined immediately, so they get no retries. Mint and deposit call external services that fail transiently (rate limits, timeouts, brief outages), so they retry with backoff. Reconcile reads back state that may be eventually consistent, so it retries a few times before flagging a mismatch for review. Every row below is a real policy this flow ships with; there are no placeholders.

Task Retries Retry delay Retry when Action on final failure
parse 0 — never (permanent fault) quarantine the raw submission
validate 0 — never (schema fault) quarantine with the validation error
mint_doi 5 10s, exponential transient registry error isolate dataset, leave draft to reconcile
deposit 5 10s, exponential HTTP 429 / 5xx isolate dataset → dead-letter store
reconcile 3 30s mismatch or read timeout alert and flag for manual review

Two invariants make these retries safe. mint_doi reserves a draft identifier and publishes it only after deposit succeeds, so a retried mint reuses the same draft rather than minting a second one. deposit is conditional on a content hash, so a retried deposit that already landed returns the existing record instead of creating a duplicate. Retries without those invariants would corrupt state; the orchestrator supplies the retry loop, but idempotency is the task’s responsibility.

Production Implementation

The flow below is complete and runnable against fake clients; swap the _FakeDataCite and _FakeRepo for real ones from the DOI Minting with DataCite and Repository Deposit Automation guides. Concurrency limits are declared with task tags and enforced by Prefect’s global concurrency limits, so the flow never exceeds the upstream API ceilings even when the batch is large.

python
from __future__ import annotations

import hashlib
from datetime import datetime, timezone

from prefect import flow, get_run_logger, task
from pydantic import BaseModel, Field, ValidationError, field_validator


class DatasetRecord(BaseModel):
    """Validated, compliance-checked representation of one submission."""

    dataset_id: str = Field(min_length=1)
    title: str = Field(min_length=3)
    creators: list[str] = Field(min_length=1)
    license_spdx: str = Field(pattern=r"^[A-Za-z0-9.\-]+$")
    created_at: datetime

    @field_validator("created_at")
    @classmethod
    def _require_utc(cls, v: datetime) -> datetime:
        if v.tzinfo is None:
            raise ValueError("created_at must be timezone-aware")
        return v.astimezone(timezone.utc)

    def content_hash(self) -> str:
        """Deterministic key: makes deposit idempotent under retry."""
        return hashlib.sha256(self.model_dump_json().encode("utf-8")).hexdigest()


class QuarantineError(RuntimeError):
    """Permanent, record-level failure routed to the dead-letter store."""


# ---- tasks -------------------------------------------------------------------

@task(retries=0)
def parse(raw: dict[str, object]) -> dict[str, object]:
    """Extract fields from a raw submission. Malformed input is permanent."""
    if not isinstance(raw, dict) or "dataset_id" not in raw:
        raise QuarantineError("unparseable submission")
    return raw


@task(retries=0)
def validate(parsed: dict[str, object]) -> DatasetRecord:
    """Enforce the typed contract; a schema failure is permanent."""
    try:
        return DatasetRecord.model_validate(parsed)
    except ValidationError as exc:
        raise QuarantineError(f"schema validation failed: {exc.error_count()} errors") from exc


@task(retries=5, retry_delay_seconds=[10, 20, 40, 80, 160], tags=["datacite"])
def mint_doi(record: DatasetRecord, client: "_FakeDataCite") -> str:
    """Reserve a draft DOI. Idempotent: same dataset_id reuses its draft."""
    logger = get_run_logger()
    doi = client.reserve_draft(record.dataset_id, title=record.title)
    logger.info("reserved draft DOI %s for %s", doi, record.dataset_id)
    return doi


@task(retries=5, retry_delay_seconds=10, tags=["repository"])
def deposit(record: DatasetRecord, doi: str, repo: "_FakeRepo") -> str:
    """Deposit conditionally on the content hash. Idempotent under retry."""
    return repo.deposit(record.content_hash(), doi)


@task(retries=3, retry_delay_seconds=30)
def reconcile(record_id: str, doi: str, repo: "_FakeRepo", client: "_FakeDataCite") -> bool:
    """Publish the draft DOI now the object is stored, and verify state."""
    stored = repo.get(record_id)
    client.publish(doi)                      # draft -> findable, only after deposit
    if stored is None or stored.doi != doi:
        raise ValueError(f"reconcile mismatch for {record_id}")
    return True


# ---- flow --------------------------------------------------------------------

@flow(name="fair-deposit")
def deposit_flow(
    submissions: list[dict[str, object]],
    client: "_FakeDataCite",
    repo: "_FakeRepo",
) -> dict[str, list[str]]:
    """Drive each submission through the chain, isolating per-dataset failures."""
    logger = get_run_logger()
    deposited: list[str] = []
    quarantined: list[str] = []

    for raw in submissions:
        dataset_id = str(raw.get("dataset_id", "<unknown>"))
        try:
            parsed = parse(raw)
            record = validate(parsed)
            doi = mint_doi(record, client)
            record_id = deposit(record, doi, repo)
            reconcile(record_id, doi, repo, client)
            deposited.append(dataset_id)
        except Exception as exc:
            # One dataset's failure must never abort the batch.
            logger.error("quarantining %s: %s", dataset_id, exc)
            repo.dead_letter(dataset_id, reason=str(exc))
            quarantined.append(dataset_id)

    logger.info("deposited=%d quarantined=%d", len(deposited), len(quarantined))
    return {"deposited": deposited, "quarantined": quarantined}

Three design choices carry the guarantees. Wrapping each submission’s chain in try/except inside the flow is what isolates failures per dataset — a raised QuarantineError or an exhausted-retry exception routes that one record to the dead-letter store while the loop continues. The tags=["datacite"] and tags=["repository"] markers bind each task to a global concurrency limit so the flow respects the upstream rate ceilings; you create the limits once with the CLI, for example prefect concurrency-limit create datacite 2 and prefect concurrency-limit create repository 4, and Prefect throttles those tagged tasks across all runs. The content_hash() and draft-then-publish sequence make retries idempotent, so Prefect’s retry loop is safe.

Note the ordering of mint and publish, because it is what makes a crash recoverable. mint_doi only reserves a draft identifier; the identifier is not made findable until reconcile publishes it, and reconcile runs after deposit has durably stored the object. If the flow dies between mint and deposit, the worst outcome is an inert draft identifier that a later reconciliation sweep can either complete or clean up — never a live, citable DOI pointing at nothing. This two-phase sequence is why the retry policies differ so sharply across the row: parse and validate guard against permanent input faults and so fail closed with no retries, while the external calls ride out transient faults with backoff, and the whole chain stays safe to re-run because no single task commits an irreversible side effect before its predecessor has succeeded. Running this flow with a Prefect backend also surfaces each of these states in the UI, so an operator sees exactly which dataset is retrying, which was quarantined, and why, without reading raw logs.

Verification

Test the flow logic without a Prefect server by running the flow in-process against fake clients. The check below asserts that a good submission is deposited exactly once and that a malformed one is quarantined without aborting the batch.

python
from dataclasses import dataclass, field


@dataclass
class _Stored:
    doi: str


@dataclass
class _FakeRepo:
    records: dict[str, _Stored] = field(default_factory=dict)
    by_hash: dict[str, str] = field(default_factory=dict)
    dead: list[str] = field(default_factory=list)

    def deposit(self, content_hash: str, doi: str) -> str:
        if content_hash in self.by_hash:
            return self.by_hash[content_hash]        # idempotent
        record_id = f"rec-{len(self.records) + 1}"
        self.records[record_id] = _Stored(doi=doi)
        self.by_hash[content_hash] = record_id
        return record_id

    def get(self, record_id: str) -> _Stored | None:
        return self.records.get(record_id)

    def dead_letter(self, dataset_id: str, reason: str) -> None:
        self.dead.append(dataset_id)


@dataclass
class _FakeDataCite:
    drafts: dict[str, str] = field(default_factory=dict)

    def reserve_draft(self, dataset_id: str, title: str) -> str:
        return self.drafts.setdefault(dataset_id, f"10.5072/{dataset_id}")  # idempotent

    def publish(self, doi: str) -> None:
        return None


def test_flow_deposits_good_and_quarantines_bad() -> None:
    good = {"dataset_id": "DS-1", "title": "Reef series", "creators": ["A. Rao"],
            "license_spdx": "CC-BY-4.0", "created_at": "2026-07-16T00:00:00+00:00"}
    bad = {"nope": True}                               # unparseable
    repo, client = _FakeRepo(), _FakeDataCite()

    result = deposit_flow(submissions=[good, bad], client=client, repo=repo)

    assert result["deposited"] == ["DS-1"]
    assert result["quarantined"] == ["<unknown>"]
    assert len(repo.records) == 1                      # deposited exactly once
    assert repo.dead == ["<unknown>"]

Run it with pytest -q. A Prefect @flow is callable in-process, so the test drives the real orchestration logic — task chaining and per-dataset isolation — without a server, scheduler, or database. A green run proves the batch survives a bad record and deposits the good one exactly once.

Gotchas

  • Retrying non-idempotent tasks. A mint_doi or deposit that is not idempotent will double-mint or double-deposit on retry. Fix: reserve-then-publish the identifier and make deposit conditional on a content hash, as above, so a retry reuses existing state.
  • Letting one failure abort the batch. Calling the tasks without a per-dataset try/except lets a single raised exception fail the whole flow run. Fix: wrap each submission’s chain and route its failure to the dead-letter store, keeping the loop going.
  • Unbounded concurrency against rate-limited APIs. Fanning out a large batch with no concurrency cap triggers HTTP 429 storms that your own retries then amplify. Fix: tag the external tasks and create Prefect global concurrency limits sized below the DataCite and repository ceilings.