Mapping Catalog Fields to Retrieval Filters

import logging
from dataclasses import dataclass
from datetime import date
from typing import Optional

log = logging.getLogger("catalog_filters")

LICENCE_CLASSES = frozenset({"open", "attribution", "share-alike", "restricted"})
PROCESSING_LEVELS = frozenset({"raw", "calibrated", "analysis-ready", "derived"})

MAX_REGION_SPAN_DEG = 60.0        # beyond this the filter is not narrowing anything


class FilterError(ValueError):
    """The requested filter cannot be satisfied — surface the reason, do not query."""


@dataclass(frozen=True)
class CatalogFilter:
    west: float
    south: float
    east: float
    north: float
    since: Optional[date] = None
    max_resolution_m: Optional[float] = None
    licence_classes: frozenset[str] = frozenset({"open", "attribution"})
    processing_level: Optional[str] = None


def validate(f: CatalogFilter) -> CatalogFilter:
    """Normalise and reject impossible filters before they reach the catalog."""
    if not (f.west < f.east and f.south < f.north):
        raise FilterError("region has zero or negative extent")
    if (f.east - f.west) > MAX_REGION_SPAN_DEG or (f.north - f.south) > MAX_REGION_SPAN_DEG:
        log.info("region spans %.1f degrees — treating as unfiltered",
                 max(f.east - f.west, f.north - f.south))
    if f.max_resolution_m is not None and f.max_resolution_m <= 0:
        raise FilterError("max_resolution_m must be a positive number of metres")
    unknown = f.licence_classes - LICENCE_CLASSES
    if unknown:
        raise FilterError(f"unknown licence class(es): {sorted(unknown)}")
    if not f.licence_classes:
        raise FilterError("at least one licence class must be permitted")
    if f.processing_level is not None and f.processing_level not in PROCESSING_LEVELS:
        raise FilterError(f"unknown processing level: {f.processing_level!r}")
    if f.since is not None and f.since.year < 1800:
        raise FilterError("since date predates any plausible catalog record")
    return f


def to_sql_params(f: CatalogFilter) -> dict:
    """Bind parameters for the catalog query; never string-format a filter into SQL."""
    f = validate(f)
    return {
        "west": f.west, "south": f.south, "east": f.east, "north": f.north,
        "since": f.since,
        "max_res": f.max_resolution_m,
        "licences": tuple(sorted(f.licence_classes)),
        "level": f.processing_level,
    }

SELECT collection_id, title, resolution_m, licence, version
FROM   catalog_collections
WHERE  geom && ST_MakeEnvelope(:west, :south, :east, :north, 4326)
  AND  ST_Intersects(geom, ST_MakeEnvelope(:west, :south, :east, :north, 4326))
  AND  (:since IS NULL OR upper(period) >= :since)
  AND  (:max_res IS NULL OR resolution_m IS NULL OR resolution_m <= :max_res)
  AND  licence_class = ANY(:licences)
  AND  (:level IS NULL OR processing_level = :level)
ORDER  BY resolution_m NULLS LAST
LIMIT  20;

from datetime import date


def test_impossible_filter_raises_with_a_reason():
    try:
        validate(CatalogFilter(-3.0, 55.0, -3.5, 56.0))     # east west of west
    except FilterError as exc:
        assert "extent" in str(exc)
        return
    raise AssertionError("an inverted region must be rejected")


def test_unknown_licence_class_is_rejected_not_ignored():
    try:
        validate(CatalogFilter(-3.5, 55.0, -3.0, 56.0,
                               licence_classes=frozenset({"public-domain-ish"})))
    except FilterError:
        return
    raise AssertionError("an unknown licence class must not silently pass")


def test_undeclared_resolution_survives_the_filter(conn):
    rows = run_catalog_query(conn, to_sql_params(
        CatalogFilter(-3.5, 55.0, -3.0, 56.0, max_resolution_m=5.0)))
    assert any(r["resolution_m"] is None for r in rows)


def test_defaults_exclude_restricted_licences():
    f = CatalogFilter(-3.5, 55.0, -3.0, 56.0)
    assert "restricted" not in f.licence_classes