Validating Code List Values Against INSPIRE Registers Jump to heading#

# sync/registers.py — requests >= 2.31 — Python 3.10+ (BUILD TIME ONLY)
import requests

TIMEOUT = 30


def fetch_register(uri: str) -> dict:
    """Fetch the machine-readable representation of a code list."""
    response = requests.get(
        uri + ".en.json",
        headers={"Accept": "application/json"},
        timeout=TIMEOUT,
    )
    response.raise_for_status()
    payload = response.json()
    # The registry reports the version of the register itself, not of each member.
    version = payload["register"]["versionInfo"]
    return {"uri": uri, "version": version, "raw": payload}

# sync/registers.py — Python 3.10+ (BUILD TIME ONLY)
import hashlib
import json
from pathlib import Path


def normalize(payload: dict) -> dict:
    """Flatten the registry representation into the stable shape validation reads."""
    members = []
    for item in payload["raw"]["register"]["containeditems"]:
        value = item["value"]
        members.append({
            "id":       value["id"].rsplit("/", 1)[-1],   # the code as it appears in data
            "uri":      value["id"],
            "label":    value["label"]["text"],
            "status":   value["status"]["label"]["text"].lower(),   # valid | superseded | retired
            "parent":   (value.get("parent") or {}).get("id", "").rsplit("/", 1)[-1] or None,
        })
    members.sort(key=lambda m: m["id"])          # stable order → stable diff
    body = {
        "uri": payload["uri"],
        "version": payload["version"],
        "members": members,
    }
    body["content_hash"] = hashlib.sha256(
        json.dumps(body, sort_keys=True, separators=(",", ":")).encode("utf-8")
    ).hexdigest()
    return body


def write_cache(body: dict, path: Path) -> None:
    path.write_text(json.dumps(body, indent=2, sort_keys=True) + "\n", encoding="utf-8")

# quality/registers.py — Python 3.10+ (RUNTIME — no network imports)
import json
from pathlib import Path


def load_pinned(spec: dict) -> tuple[frozenset[str], frozenset[str], dict[str, str | None]]:
    path = Path(spec["cache_path"])
    if not path.is_file():
        raise FileNotFoundError(
            f"{spec['name']}: register cache {path} is missing. Run the sync job; "
            "validation never fetches."
        )
    body = json.loads(path.read_text(encoding="utf-8"))

    if body["version"] != spec["register_version"]:
        raise ValueError(
            f"{spec['name']}: cache holds {body['version']}, manifest pins "
            f"{spec['register_version']} — refusing to validate."
        )
    if body["uri"] != spec["register_uri"]:
        raise ValueError(f"{spec['name']}: cache is for {body['uri']}, not {spec['register_uri']}")

    active = frozenset(m["id"] for m in body["members"] if m["status"] == "valid")
    retired = frozenset(m["id"] for m in body["members"] if m["status"] != "valid")
    parents = {m["id"]: m["parent"] for m in body["members"]}
    return active, retired, parents

# quality/registers.py — Python 3.10+
def resolve(value: str, active: frozenset[str], retired: frozenset[str],
            parents: dict[str, str | None], spec: dict) -> tuple[bool, str]:
    if value in active:
        return True, ""
    if value in retired:
        if spec.get("allow_deprecated"):
            return True, "ATTR_DEPRECATED_CODE_ACCEPTED"
        return False, "ATTR_DEPRECATED_CODE"

    if spec.get("allow_narrower"):
        # Walk up the hierarchy: a narrower term is acceptable if an ancestor is active.
        seen: set[str] = set()
        cursor = parents.get(value)
        while cursor and cursor not in seen:
            if cursor in active:
                return True, "ATTR_NARROWER_TERM_ACCEPTED"
            seen.add(cursor)
            cursor = parents.get(cursor)

    return False, "ATTR_UNDEFINED_CODE"

# .github/workflows/sync-registers.yml
name: sync-inspire-registers
on:
  schedule:
    - cron: "0 4 * * 1"        # weekly, Monday 04:00 UTC
  workflow_dispatch:

jobs:
  sync:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: {python-version: "3.11"}
      - run: pip install "requests>=2.31"
      - run: python -m sync.registers --manifest attribute_domains.yaml --out registers/
      - name: Report what would change
        run: python -m sync.registers --impact --input data/latest_delivery.parquet
      - uses: peter-evans/create-pull-request@v6
        with:
          branch: chore/register-sync
          title: "Register sync: new code list version(s) available"
          body-path: registers/IMPACT.md

# pytest >= 7
def test_pinned_cache_matches_manifest():
    active, retired, parents = load_pinned(LAND_USE_SPEC)
    assert "1_1_1" in active
    assert len(active) == 62                    # the pinned version's member count

def test_stale_cache_refuses_to_validate(tmp_path):
    spec = dict(LAND_USE_SPEC, register_version="1999-01-01")
    with pytest.raises(ValueError, match="refusing to validate"):
        load_pinned(spec)

def test_validation_path_never_imports_requests():
    # Negative control: proves the runtime cannot reach the network.
    import sys
    import quality.registers          # noqa: F401
    assert "requests" not in sys.modules

# jq >= 1.6
jq -r '"\(.uri)  version=\(.version)  members=\(.members | length)"' registers/hilucs-2023-04-11.json
# http://inspire.ec.europa.eu/codelist/HILUCSValue  version=2023-04-11  members=62