Validating Metadata Against FAIR Criteria Automatically: A Python Implementation Guide #

from __future__ import annotations

import asyncio
import re
from dataclasses import dataclass, field
from enum import Enum
from typing import Any

import httpx

# --- Controlled vocabularies (in production, load from a versioned registry) ---
SPDX_LICENSES: frozenset[str] = frozenset(
    {"CC-BY-4.0", "CC0-1.0", "CC-BY-SA-4.0", "MIT", "Apache-2.0"}
)
DATACITE_RESOURCE_TYPES: frozenset[str] = frozenset(
    {"Dataset", "Software", "Collection", "Text", "Image"}
)
ACCESS_RIGHTS: frozenset[str] = frozenset({"open", "embargoed", "restricted"})

PID_SHAPE = re.compile(r"^(10\.\d{4,9}/\S+|hdl:\S+|ark:/\d+/\S+)$")
ORCID_URI = re.compile(r"^https://orcid\.org/\d{4}-\d{4}-\d{4}-\d{3}[\dX]$")


class Severity(str, Enum):
    CRITICAL = "critical"
    WARNING = "warning"


@dataclass(slots=True)
class Finding:
    rule_id: str
    principle: str
    severity: Severity
    message: str


@dataclass(slots=True)
class FairReport:
    content_hash: str
    findings: list[Finding] = field(default_factory=list)

    @property
    def blocked(self) -> bool:
        """A record is blocked only by a CRITICAL finding."""
        return any(f.severity is Severity.CRITICAL for f in self.findings)

    @property
    def score(self) -> float:
        """Fraction of the 10 rules that passed; warnings still count as misses."""
        return round(1.0 - len(self.findings) / 10, 3)

    @property
    def verdict(self) -> str:
        if self.blocked:
            return "blocked"
        return "deferred" if self.findings else "compliant"


def _structural_checks(record: dict[str, Any], report: FairReport) -> None:
    """Cheap, offline checks. Populate report.findings with structural gaps."""
    def fail(rule: str, principle: str, sev: Severity, msg: str) -> None:
        report.findings.append(Finding(rule, principle, sev, msg))

    pid = str(record.get("identifier", "")).strip()
    if not PID_SHAPE.match(pid):
        fail("F1-pid-present", "Findable", Severity.CRITICAL,
             f"identifier {pid!r} is not a DOI/Handle/ARK")
    if not record.get("title") or not record.get("creators"):
        fail("F2-rich-metadata", "Findable", Severity.CRITICAL,
             "title and at least one creator are required")
    access_url = str(record.get("accessUrl", ""))
    if not access_url.startswith("https://"):
        fail("A1-access-protocol", "Accessible", Severity.CRITICAL,
             "accessUrl must be an https URL")
    if record.get("accessRights") not in ACCESS_RIGHTS:
        fail("A1-access-rights", "Accessible", Severity.CRITICAL,
             f"accessRights must be one of {sorted(ACCESS_RIGHTS)}")
    if not str(record.get("@context", "")).startswith("http"):
        fail("I1-context", "Interoperable", Severity.CRITICAL,
             "a JSON-LD @context URI is required")
    if record.get("resourceTypeGeneral") not in DATACITE_RESOURCE_TYPES:
        fail("I2-resource-type", "Interoperable", Severity.WARNING,
             "resourceTypeGeneral is outside the DataCite vocabulary")
    if record.get("rights") not in SPDX_LICENSES:
        fail("R1-license-spdx", "Reusable", Severity.CRITICAL,
             f"rights {record.get('rights')!r} is not an SPDX identifier")
    if not record.get("provenance"):
        fail("R1-provenance", "Reusable", Severity.WARNING,
             "provenance block (source + timestamp) is missing")


async def _resolves(client: httpx.AsyncClient, url: str) -> bool:
    """A timeout or transport error is a soft miss, never an exception."""
    try:
        resp = await client.get(url, follow_redirects=True, timeout=5.0)
        return resp.status_code == 200
    except httpx.HTTPError:
        return False


async def _resolution_checks(
    record: dict[str, Any], report: FairReport, client: httpx.AsyncClient
) -> None:
    """Concurrent external checks; failures downgrade to WARNING (deferred)."""
    sem = asyncio.Semaphore(8)  # bound concurrent outbound calls

    async def guarded(url: str) -> bool:
        async with sem:
            return await _resolves(client, url)

    pid = str(record.get("identifier", "")).strip()
    creators = record.get("creators") or []
    orcids = [c.get("orcid", "") for c in creators if ORCID_URI.match(c.get("orcid", ""))]

    pid_ok, *orcid_oks = await asyncio.gather(
        guarded(f"https://doi.org/{pid}"),
        *[guarded(o) for o in orcids],
    )
    if not pid_ok:
        report.findings.append(Finding(
            "F1-pid-resolves", "Findable", Severity.WARNING,
            f"identifier {pid} did not resolve; deferred for retry"))
    if len(orcids) < len(creators) or not all(orcid_oks):
        report.findings.append(Finding(
            "R1-creator-orcid", "Reusable", Severity.WARNING,
            "one or more creator ORCIDs are missing or unresolved"))


async def validate_fair(
    record: dict[str, Any], content_hash: str
) -> FairReport:
    """Run the full FAIR gate: structural first, then bounded async resolution."""
    report = FairReport(content_hash=content_hash)
    _structural_checks(record, report)
    # Short-circuit: never spend network budget on a structurally broken record.
    if report.blocked:
        return report
    async with httpx.AsyncClient() as client:
        await _resolution_checks(record, report, client)
    return report

import asyncio


def _record(**overrides: object) -> dict[str, object]:
    base = {
        "identifier": "10.5281/zenodo.123456",
        "title": "Soil Carbon Flux, 2026",
        "creators": [{"orcid": "https://orcid.org/0000-0002-1825-0097"}],
        "accessUrl": "https://example.org/ds/1",
        "accessRights": "open",
        "@context": "https://schema.org/",
        "resourceTypeGeneral": "Dataset",
        "rights": "CC-BY-4.0",
        "provenance": {"source": "eln", "at": "2026-05-01T09:30:00Z"},
    }
    return base | overrides


def test_verdicts(monkeypatch) -> None:
    # Force every external resolution to succeed for a deterministic test.
    async def always_ok(client, url):  # noqa: ANN001
        return True
    monkeypatch.setattr("mymod._resolves", always_ok)

    clean = asyncio.run(validate_fair(_record(), "h1"))
    assert clean.verdict == "compliant" and clean.score == 1.0

    bad = asyncio.run(validate_fair(_record(rights="Open Access"), "h2"))
    assert bad.verdict == "blocked"            # non-SPDX license is critical

    gap = asyncio.run(validate_fair(_record(provenance=None), "h3"))
    assert gap.verdict == "deferred"           # missing provenance is only a warning