Start at the landing page, read /conformance to learn which capabilities are real, list the collections, fetch the catalogue’s queryables, then search /collections/{id}/items with q, bbox, datetime and a CQL2 filter. The one non-optional step is the conformance check: a service without the filter class ignores your filter and returns the entire catalogue with a 200.
CSW fails toward emptiness — a wrong typeNames or outputSchema yields zero records, which is alarming and gets investigated. A Records service fails toward completeness: an unimplemented filter parameter is ignored, and the response is a valid page of records that simply were not filtered.
A client that trusts it processes the whole catalogue believing it processed a narrow slice. In a harvesting pipeline that means ingesting thousands of irrelevant records; in a user interface it means a search box that appears to do nothing. Neither surfaces as an error, which is why the discovery walk above ends with an assertion rather than a request.
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any, Iterator
import requests
CONF_FILTER = "http://www.opengis.net/spec/ogcapi-records-1/1.0/conf/filter"
CONF_CQL2_TEXT = "http://www.opengis.net/spec/cql2/1.0/conf/cql2-text"
CONF_SORTING = "http://www.opengis.net/spec/ogcapi-records-1/1.0/conf/sorting"
class CapabilityMissing(RuntimeError):
"""Raised instead of letting an ignored parameter return everything."""
@dataclass
class RecordHit:
identifier: str
title: str
description: str
updated: str | None
keywords: list[str] = field(default_factory=list)
links: list[dict[str, Any]] = field(default_factory=list)
def data_link(self) -> str | None:
"""Where the described resource actually lives, if it says."""
for link in self.links:
if link.get("rel") in {"item", "enclosure", "service"}:
return link.get("href")
return None
@classmethod
def from_feature(cls, feature: dict[str, Any]) -> "RecordHit":
props = feature.get("properties") or {}
return cls(
identifier=str(feature.get("id", "")),
title=props.get("title") or "",
description=props.get("description") or "",
updated=props.get("updated"),
keywords=list(props.get("keywords") or []),
links=list(feature.get("links") or []),
)
@dataclass
class RecordsClient:
base: str
session: requests.Session = field(default_factory=requests.Session)
timeout: int = 60
_conformance: set[str] | None = field(default=None, init=False)
def _get(self, url: str, **params: Any) -> dict[str, Any]:
resp = self.session.get(url, params=params or None, timeout=self.timeout,
headers={"Accept": "application/json"})
resp.raise_for_status()
return resp.json()
def landing(self) -> dict[str, Any]:
return self._get(self.base.rstrip("/"))
def conformance(self) -> set[str]:
if self._conformance is None:
payload = self._get(f"{self.base.rstrip('/')}/conformance")
self._conformance = set(payload.get("conformsTo", []))
return self._conformance
def collections(self) -> list[dict[str, Any]]:
payload = self._get(f"{self.base.rstrip('/')}/collections")
return payload.get("collections", [])
def queryables(self, collection: str) -> set[str]:
try:
payload = self._get(
f"{self.base.rstrip('/')}/collections/{collection}/queryables")
except requests.HTTPError:
return set()
return set((payload.get("properties") or {}).keys())
def search(self, collection: str, *, q: str | None = None,
bbox: tuple[float, float, float, float] | None = None,
datetime_range: str | None = None,
cql2: str | None = None, limit: int = 100,
max_records: int | None = None) -> Iterator[RecordHit]:
"""Search a catalogue, following next links until exhausted.
Refuses to send a CQL2 filter the service has not declared support
for — an ignored filter returns the whole catalogue silently.
"""
if cql2:
conf = self.conformance()
missing = {CONF_FILTER, CONF_CQL2_TEXT} - conf
if missing:
raise CapabilityMissing(
f"{self.base} does not declare {sorted(missing)}; a filter "
"would be ignored and every record returned")
params: dict[str, Any] = {"limit": limit}
if q:
params["q"] = q
if bbox:
params["bbox"] = ",".join(f"{v:g}" for v in bbox)
if datetime_range:
params["datetime"] = datetime_range
if cql2:
params["filter"] = cql2
params["filter-lang"] = "cql2-text"
url = f"{self.base.rstrip('/')}/collections/{collection}/items"
seen_urls: set[str] = set()
yielded = 0
while url:
payload = self._get(url, **params)
params = {} # next hrefs are already complete
for feature in payload.get("features", []):
yield RecordHit.from_feature(feature)
yielded += 1
if max_records and yielded >= max_records:
return
nxt = next((l["href"] for l in payload.get("links", [])
if l.get("rel") == "next"), None)
if nxt and nxt in seen_urls:
raise RuntimeError(f"paging loop: {nxt} repeated")
if nxt:
seen_urls.add(nxt)
url = nxt
if __name__ == "__main__":
client = RecordsClient("https://example.org/records")
catalogue = client.collections()[0]["id"]
print("queryable:", sorted(client.queryables(catalogue))[:8])
for hit in client.search(catalogue, q="flood", bbox=(5.9, 45.8, 10.5, 47.8),
datetime_range="2023-01-01/..",
cql2="type = 'dataset'", max_records=20):
print(f"{hit.identifier:<40} {hit.title[:50]:<52} {hit.data_link()}")
Conformance is checked before the filter is sent, not after. search raises CapabilityMissing naming the classes the service failed to declare. The alternative — sending the filter and hoping — produces a full catalogue dump that looks like a successful narrow search, which is the failure this whole client is shaped around.
Constraints combine with AND, always. Free text, bounding box, datetime, type and CQL2 can all be sent together and the service intersects them. There is no parameter syntax for OR across them; anything more complex belongs entirely inside the CQL2 expression, which does have boolean composition.
Parameters are dropped after the first page. A rel="next" href is a complete URL that already carries the query, so re-appending the original parameters either duplicates them or, worse, overrides the server’s cursor. Clearing params after the first request is a two-character detail that prevents a paging loop.
Repeated next hrefs are a server bug, not a stop condition. A service that returns the same next link twice will spin a naive client forever. Tracking seen URLs and raising turns an infinite loop into a diagnosable error — the same guard the Features paging guide recommends.
data_link is what makes a hit useful. A record with no item, enclosure or service link describes something the client cannot reach. Surfacing it as None rather than digging through the links array at every call site makes the gap visible.
Confirm the discovery walk and that filtering actually narrows:
client = RecordsClient("https://example.org/records")
print(sorted(c for c in client.conformance() if "records" in c))
catalogue = client.collections()[0]["id"]
broad = sum(1 for _ in client.search(catalogue, limit=500, max_records=5000))
narrow = sum(1 for _ in client.search(catalogue, cql2="type = 'service'",
limit=500, max_records=5000))
print(broad, "->", narrow)
assert narrow < broad, "the filter changed nothing — is it being applied?"
['http://www.opengis.net/spec/ogcapi-records-1/1.0/conf/core',
'http://www.opengis.net/spec/ogcapi-records-1/1.0/conf/filter',
'http://www.opengis.net/spec/ogcapi-records-1/1.0/conf/sorting']
1284 -> 37
A filter that leaves the count unchanged is the signature of the superset failure, and asserting on it costs one extra query.
A bounding box matches extents, not footprints. Records usually carry a rectangular extent, so a national dataset’s record intersects essentially every bounding box a user draws. That is correct and routinely reported as a bug. Where precision matters, filter additionally on a resolution or scale property if the catalogue publishes one.
datetime filters the record’s temporal extent, not its update time. A search for records about 2024 and a search for records changed in 2024 are different queries; the second uses a CQL2 predicate on the updated property. Harvesters want the second, which is the watermark discipline applied to a new protocol.
Free-text q is not defined precisely. The specification leaves the matching semantics to the implementation, so one service does prefix matching on the title and another does stemmed full-text over every field. Never assume a q result set is reproducible across services; use CQL2 where the query must be exact.
Sorting is a separate conformance class. Requesting sortby against a service that has not declared it is ignored, producing results in whatever order the store returns — which is stable enough to be misleading and unstable enough to break paging assumptions.
Because an unimplemented filter is ignored, not rejected. The response is a valid page of unfiltered records with a 200 status, so a client that trusts it processes the whole catalogue believing it processed a slice. Checking one document once per session turns that into an exception naming the missing class.
CQL2 whenever the query must be exact or reproducible. The free-text q parameter has implementation-defined semantics, so the same query against two catalogues returns different sets — fine for a search box, unusable for a pipeline that must select a specific set of records.
Filter on the updated property with a CQL2 comparison against your watermark, page by following next links, and advance the watermark only after the whole run completes. That is exactly the CSW incremental pattern with different request syntax; the discipline that makes it safe is unchanged.
Treat every property as unfilterable and rely on the standard parameters — q, bbox, datetime and type — which are part of the core. Sending a CQL2 predicate on an undeclared property is silently ignored, which returns a superset, so guessing is worse than not filtering.
Back to OGC API - Records and Catalog Modernisation
Related