Harvest the existing catalogue through CSW into one normalised record model, serve that model as OGC API - Records, and keep CSW as a second serialiser over the same store rather than a second catalogue. Preserve the dc:identifier as the Feature id, carry every CI_OnlineResource into the links array, and gate the cutover on both protocols returning identical identifier sets for the same query.
Mapping ISO 19139 onto a Records property object is mostly clerical. Title, abstract, keywords and extent sit in predictable places and convert without judgement. The exception is linkage, which lives in gmd:CI_OnlineResource elements nested inside gmd:MD_Distribution inside gmd:distributionInfo — several levels below anything else being read.
A conversion that stops at the descriptive fields produces records that validate, look complete in a listing, and point nowhere. Because nothing in the Records specification requires a link, no validator flags it, and the defect surfaces only when a user clicks through and finds nothing to click. That is why the equivalence tests below check link presence explicitly rather than only record counts.
from __future__ import annotations
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any, Iterator
from xml.etree import ElementTree as ET
NS = {
"gmd": "http://www.isotc211.org/2005/gmd",
"gco": "http://www.isotc211.org/2005/gco",
"csw": "http://www.opengis.net/cat/csw/2.0.2",
}
# ISO protocol tokens mapped onto IANA-ish link relations. Anything not
# recognised still becomes a link, tagged rel="related" — dropping it is
# the failure this whole module exists to prevent.
PROTOCOL_REL = {
"OGC:WMS": ("service", "WMS"),
"OGC:WFS": ("service", "WFS"),
"OGC:WCS": ("service", "WCS"),
"WWW:DOWNLOAD-1.0-http--download": ("enclosure", None),
"WWW:LINK-1.0-http--link": ("related", None),
}
@dataclass
class Link:
href: str
rel: str
title: str | None = None
type: str | None = None
def as_json(self) -> dict[str, Any]:
return {k: v for k, v in
{"href": self.href, "rel": self.rel,
"title": self.title, "type": self.type}.items() if v}
@dataclass
class Record:
identifier: str
title: str
description: str
updated: datetime
bbox: tuple[float, float, float, float] | None = None
keywords: list[str] = field(default_factory=list)
links: list[Link] = field(default_factory=list)
record_type: str = "dataset"
def as_feature(self) -> dict[str, Any]:
geometry = None
if self.bbox:
w, s, e, n = self.bbox
geometry = {"type": "Polygon", "coordinates": [[
[w, s], [e, s], [e, n], [w, n], [w, s]]]}
return {
"type": "Feature",
"id": self.identifier, # never reminted
"geometry": geometry,
"properties": {
"type": self.record_type,
"title": self.title,
"description": self.description,
"updated": self.updated.isoformat(),
"keywords": list(self.keywords),
},
"links": [link.as_json() for link in self.links],
}
def _text(node: ET.Element | None, path: str) -> str:
if node is None:
return ""
found = node.findtext(f"{path}/gco:CharacterString", "", NS)
return (found or "").strip()
def from_iso(md: ET.Element) -> Record:
"""Convert one gmd:MD_Metadata element into the record model."""
identification = md.find(".//gmd:identificationInfo//"
"gmd:MD_DataIdentification", NS)
citation = (identification.find(".//gmd:citation//gmd:CI_Citation", NS)
if identification is not None else None)
bbox = None
box = md.find(".//gmd:EX_GeographicBoundingBox", NS)
if box is not None:
def ordinate(tag: str) -> float:
return float(box.findtext(f"gmd:{tag}/gco:Decimal", "0", NS))
bbox = (ordinate("westBoundLongitude"), ordinate("southBoundLatitude"),
ordinate("eastBoundLongitude"), ordinate("northBoundLatitude"))
links: list[Link] = []
for online in md.findall(".//gmd:CI_OnlineResource", NS):
href = _text(online, "gmd:linkage").replace("gmd:URL", "").strip()
if not href:
href = (online.findtext("gmd:linkage/gmd:URL", "", NS) or "").strip()
if not href:
continue
protocol = _text(online, "gmd:protocol")
rel, type_hint = PROTOCOL_REL.get(protocol, ("related", None))
links.append(Link(href=href, rel=rel,
title=_text(online, "gmd:name") or None,
type=type_hint))
stamp = md.findtext("gmd:dateStamp/gco:DateTime", "", NS) or \
md.findtext("gmd:dateStamp/gco:Date", "", NS)
updated = (datetime.fromisoformat(stamp.replace("Z", "+00:00"))
if stamp else datetime.now(timezone.utc))
return Record(
identifier=_text(md, "gmd:fileIdentifier"),
title=_text(citation, "gmd:title"),
description=_text(identification, "gmd:abstract"),
updated=updated,
bbox=bbox,
keywords=[(k.text or "").strip()
for k in md.findall(".//gmd:MD_Keywords//gmd:keyword/"
"gco:CharacterString", NS)
if (k.text or "").strip()],
links=links,
)
def harvest(csw_url: str, page: int = 50) -> Iterator[Record]:
"""Page a CSW catalogue and yield normalised records."""
import requests
start = 1
while True:
resp = requests.get(csw_url, timeout=120, params={
"service": "CSW", "version": "2.0.2", "request": "GetRecords",
"typeNames": "gmd:MD_Metadata",
"outputSchema": "http://www.isotc211.org/2005/gmd",
"elementSetName": "full", "resultType": "results",
"startPosition": str(start), "maxRecords": str(page),
})
resp.raise_for_status()
root = ET.fromstring(resp.content)
results = root.find("csw:SearchResults", NS)
if results is None:
return
for md in results.findall("gmd:MD_Metadata", NS):
yield from_iso(md)
nxt = int(results.get("nextRecord", "0"))
if nxt <= 0 or nxt == start:
return
start = nxt
def migration_report(records: list[Record]) -> dict[str, Any]:
"""What the conversion could not carry — read this before cutting over."""
return {
"records": len(records),
"without_identifier": sum(1 for r in records if not r.identifier),
"without_links": sum(1 for r in records if not r.links),
"without_extent": sum(1 for r in records if r.bbox is None),
"without_abstract": sum(1 for r in records if not r.description),
}
Unknown protocols still become links. PROTOCOL_REL maps the ISO protocol tokens it recognises and falls back to rel="related" for everything else. That fallback is deliberate: catalogues are full of protocol values nobody standardised, and dropping a link because its protocol token is unfamiliar loses exactly the association the record exists to record.
The identifier is carried, never minted. as_feature sets the Feature id from gmd:fileIdentifier unchanged. Existing harvest states, citations and cross-references all use that string; a Records service that generates fresh identifiers breaks every one of them silently, and the breakage is only discovered by whoever depended on it.
The report is read before cutover, not after. migration_report counts what could not be carried — records with no link, no extent, no abstract — and those counts are the migration’s real status. A conversion that produces ten thousand valid records of which four thousand have no link has not migrated a catalogue; it has produced a listing.
Serve both, retire on evidence. The store and the query layer are shared, so CSW becomes a second serialiser rather than a second system. Retirement then waits on access logs showing nothing calls it — which for a catalogue with regulated harvesting obligations may be years, and is a decision for the data rather than the calendar.
Run the report, then assert the two protocols agree:
records = list(harvest("https://legacy.example.org/csw"))
print(migration_report(records))
{'records': 4182, 'without_identifier': 0, 'without_links': 137,
'without_extent': 12, 'without_abstract': 3}
One hundred and thirty-seven records with no link is the number to act on — either the source genuinely lacks linkage, or the CI_OnlineResource path is wrong for those records’ profile.
def test_protocols_agree(csw, records_client, catalogue):
query = {"keyword": "hydrology", "bbox": (5.9, 45.8, 10.5, 47.8)}
csw_ids = set(csw.search(**query))
api_ids = {hit.identifier for hit in records_client.search(
catalogue, q="hydrology", bbox=query["bbox"])}
assert csw_ids == api_ids, (
f"only in CSW: {sorted(csw_ids - api_ids)[:5]}; "
f"only in Records: {sorted(api_ids - csw_ids)[:5]}")
Profiles nest linkage differently. INSPIRE, national profiles and vendor extensions all place CI_OnlineResource at slightly different depths, which is why the code searches for it anywhere in the document rather than following one fixed path. A path-based read that works against one catalogue’s records commonly returns nothing against another’s.
Date stamps come in two flavours. gco:DateTime and gco:Date both appear as the value of gmd:dateStamp, and a parser expecting one silently produces no timestamp for the other — which then breaks incremental harvesting because every record looks equally old.
Do not migrate the model at the same time. Moving from ISO 19139 to a Records property object is a serialisation change. Changing what fields the catalogue holds, or adopting a different metadata profile, is a separate project. Doing both at once means an equivalence test can never pass, which removes the only gate the migration has.
Keep harvesting through CSW during the transition. The source of truth stays where it is until the new service is proven. The incremental discipline in Scheduling Recurring CSW Harvests With Python applies unchanged — advance the watermark only after a clean run.
For as long as something calls it, which in a regulated environment means until the national or INSPIRE harvester adopts Records. Since it becomes a serialiser over the same store rather than a separate catalogue, the ongoing cost is small — and access logs, not a roadmap, are what should decide the retirement date.
Links. They live several levels deeper than every other field, inside distribution information, and their absence breaks no validator. A record with a title, abstract, keywords and extent looks complete in any listing while pointing at nothing, which is why the migration report counts records without links as a first-class metric.
No. Existing citations, harvest states and cross-references all use the CSW identifier, and reminting breaks them all with no error anywhere. Carry gmd:fileIdentifier through as the Feature id even when it is an unattractive UUID.
When the same logical query returns the same identifier set through both protocols, and when the count of records missing links, extents or abstracts is either zero or explained by the source data. Record counts alone are not sufficient — two catalogues can hold the same number of records and disagree about which ones.
Back to OGC API - Records and Catalog Modernisation
Related