REST converters: page by id windows

.gitignore

@@ -12,6 +12,8 @@ __pycache__/
 *.egg-info/
 /.pytest_cache/
 /.ruff_cache/
+/.venv/
+uv.lock
 
 # Coverage
 .coverage

CHANGELOG.md

@@ -36,14 +36,17 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 - Updated `aiohttp` to support Zenodo responses that include both returned `Content-Type` headers.
 - Improved geometry axis handling so generated tiles and bounding boxes keep x/y order consistent in output.
 - Updated vecorel-cli to 0.2.16, 0.2.17, 0.2.18 and 0.2.20, including improved validation defaults and latest-variant selection when `--variant` is not provided.
+- REST converters download much faster from large layers and retry when a service answers with intermittent errors.
 - BE-VLG: Extended editions to 2018-2026 and aligned determination dates with the selected campaign year.
 - CZ: Extended year coverage, including GPZ_DP editions (2019-2022), and added 2026 nested-archive support.
 - DE-SH: Extended support to editions 2023, 2025 and 2026.
 - DK: Editions now cover 2008-2026. The 2008 and 2009 editions are published without the crop and HCAT extensions because the source has no crop columns.
 - ES:
   - ES regions based on SIGPAC now publish `hcat:code` from land-use mapping.
   - ES-AR now reads municipality SIGPAC sources listed by IDEAragon.
+  - ES-CB now covers editions 2010-2025.
   - ES-GA now supports editions 2014-2026.
+  - ES-IB now covers editions 2022-2026, reading the current and the historic SIGPAC services.
 - FI: Editions are now available by year (2020-2025).
 - FR: Editions now cover 2017-2024, mapped through one shared crop code list (https://fiboa.org/code/fr/fr.csv).
 - HR: Editions now cover 2011-2024.
@@ -61,7 +64,10 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 ### Fixed
 - Added HCAT spelling fixes via `csv_supplements` for DE-BB, DE-NDS and EC-SI.
 - Declared the `beautifulsoup4` dependency used by ES-PV and ES-VC.
-- Dropped cached error pages for REST converters.
+- REST converters:
+  - Downloaded data cached for one dataset, edition or service is no longer served for another. Previously cached downloads are fetched again once.
+  - Error responses and interrupted downloads are no longer cached.
+  - Layers that join several tables (some ES-CB and ES-IB editions) are now filtered and paged correctly, and their column names no longer carry table prefixes.
 - Fixed `use_variant_as_determination` so determination dates are retained.
 - Multipart geometries now get recomputed area/perimeter for split parts.
 - Rows missing `crop:code` are now dropped with a warning (and an error threshold), instead of failing whole conversions.
@@ -90,10 +96,12 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
   - ES-AN now uses the correct land-use column and campaign-based determination date.
   - ES-CAT and ES-CN now map crop codes to HCAT with the extended mapping table.
   - ES-CB now derives determination date from the campaign.
+  - ES-CB now ships Cantabria's own licence instead of CC-BY-NC, with the province as provider.
   - ES-CL now reads the HTTPS source and 2025 province subfolders.
   - ES-CM now uses the campaign-specific SIGPAC service and schema.
   - ES-CN now keeps distinct island records and correct region metadata.
   - ES-EX and ES-NC now read FEGA national recinto releases (2025, 2026) because the regional portals are unavailable.
+  - ES-IB now names the Balearic government as provider instead of Navarra's.
   - ES-MD now finds `RECINTO.shp` regardless of archive folder layout.
 - Europe-LAND: Empty source crop codes now fall back to crop names (for example LT 2024).
 - FR:

fiboa_cli/conversion/converter_rest.py

@@ -1,4 +1,7 @@
 import os
+import re
+import time
+import zlib
 from urllib.parse import urlencode
 
 import geopandas as gpd
@@ -17,12 +20,16 @@ def rest_layer_filter(self, layers):
         return next(iter(layers))
 
     def get_urls(self):
-        assert self.rest_base_url, (
-            "Either define {c}.rest_base_url or override {c}.get_urls()".format(
-                c=self.__class__.__name__
-            )
+        # An edition may live in a service of its own: es_ib keeps the current
+        # snapshot in one and the yearly ones in another, so a variant whose
+        # value is a URL names the service to read it from.
+        url = self.variants.get(self.variant or next(iter(self.variants), ""))
+        if not isinstance(url, str) or not url.startswith("http"):
+            url = self.rest_base_url
+        assert url, "Either define {c}.rest_base_url or override {c}.get_urls()".format(
+            c=self.__class__.__name__
         )
-        return {"REST": self.rest_base_url}
+        return {"REST": url}
 
     def download_files(self, uris, cache_folder=None):
         # Read-data will just stream all pages of rest-service
@@ -33,6 +40,24 @@ def download_files(self, uris, cache_folder=None):
         # This happens when input_file param is used
         return super().download_files(uris, cache_folder)
 
+    @staticmethod
+    def _unqualify(gdf):
+        """Drop the table prefix a joined layer puts on every field name.
+
+        A join answers with SIGPAC_FOGAIBA.DN_OID rather than DN_OID, so a
+        converter's `columns` match nothing. The first table wins, which is the
+        one carrying the geometry.
+        """
+        if not any("." in c for c in gdf.columns):
+            return gdf
+        renames = {}
+        for column in gdf.columns:
+            name = column.rsplit(".", 1)[-1]
+            if name not in gdf.columns and name not in renames.values():
+                renames[column] = name
+        gdf = gdf.rename(columns=renames)
+        return gdf.loc[:, ~gdf.columns.duplicated()]
+
     def get_data(self, paths, **kwargs):
         if not (isinstance(paths[0], str) and paths[0].startswith("http")):
             # This happens when the input_file param is used. Pages are read with the same
@@ -43,7 +68,7 @@ def get_data(self, paths, **kwargs):
             # map from their own attribute. Reading a fixture must match a real run.
             for path, uri in paths:
                 self.info(f"Reading {path} into GeoDataFrame")
-                yield gpd.read_file(path), path, uri, None
+                yield self._unqualify(gpd.read_file(path)), path, uri, None
             return
 
         base_url = paths[0]  # loop over paths to support more than 1 source
@@ -54,38 +79,134 @@ def get_data(self, paths, **kwargs):
         layer = self.rest_layer_filter(service_metadata["layers"])
         page_size = service_metadata["maxRecordCount"]
         layer_url = f"{base_url}/{layer['id']}/query"
-        rest_params = dict(self.rest_params)
-        base_where = rest_params.pop("where", None)  # combined with the paging filter below
-        get_dict = rest_params | {
+        # Joined layers qualify every field with the table name; discover the
+        # real key field before paging on it ("OBJECTID" alone fails there).
+        probe = requests.get(
+            layer_url,
+            {
+                "f": "json",
+                "where": "1=1",
+                "outFields": "*",
+                "resultRecordCount": 1,
+                "returnGeometry": "false",
+            },
+        ).json()
+        attribute = self.rest_attribute
+        if probe.get("features"):
+            names = list(probe["features"][0]["attributes"].keys())
+            attribute = next(
+                (n for n in names if n == self.rest_attribute),
+                next(
+                    (n for n in names if n.endswith("." + self.rest_attribute)), self.rest_attribute
+                ),
+            )
+        base_where = self.rest_params.get("where")
+
+        # Page by half-open id windows rather than orderByFields + "id > last":
+        # server-side sorting costs ~100 s per request on joined layers. The key
+        # is unique, so a window of page_size ids cannot overflow a page.
+        min_id = self._rest_id_bound(layer_url, attribute, base_where, "ASC")
+        max_id = self._rest_id_bound(layer_url, attribute, base_where, "DESC")
+
+        get_dict = self.rest_params | {
             "outFields": "*",
             "returnGeometry": "true",
             "f": self.rest_format,
-            "sortBy": self.rest_attribute,
-            "resultRecordCount": page_size,
         }
-        gdfs = []
-        last_id = -1
-        while True:
-            get_dict["where"] = f"{self.rest_attribute}>{last_id}"
-            if base_where:
-                get_dict["where"] += f" AND ({base_where})"
-            url = f"{layer_url}?{urlencode(get_dict)}"
-            if cache_fs is not None:
-                cache_file = os.path.join(
-                    cache_folder, f"{self.id}_{layer['id']}_{last_id}.{self.rest_format}"
-                )
-                if not cache_fs.exists(cache_file):
-                    with cache_fs.open(cache_file, mode="wb") as file:
-                        stream_file(source_fs, url, file)
-                url = cache_file
-
-            data = gpd.read_file(url)
-            print(
-                f"Read {len(data)} features, page {len(gdfs)} from [{data.iloc[0, 0]} ... {data.iloc[-1, 0]}]"
-            )
-            last_id = data[self.rest_attribute].values[-1]
+        # Layer ids repeat across services (every SIXPAC_<year> MapServer has its
+        # Recintos layer at id 2), so the service must be part of the cache key.
+        # So must the filter (de_st selects its edition by `where` alone, on one
+        # service and one layer) and the window's upper bound: page_size follows
+        # the service's maxRecordCount, and a page kept from a smaller one would
+        # silently drop every id above its own bound.
+        service = re.sub(r"\W+", "_", base_url.rstrip("/").split("/rest/services/")[-1])
+        where_key = f"_w{zlib.crc32(base_where.encode()):08x}" if base_where else ""
+        prefix = f"{self.id}_{service}_{layer['id']}{where_key}_r"
+        windows = self._cached_windows(cache_fs, cache_folder, prefix)
+        page = 0
+        lo = min_id - 1
+        while lo < max_id:
+            cached = lo in windows
+            hi = windows[lo] if cached else lo + page_size
+            if cached:
+                url = os.path.join(cache_folder, f"{prefix}{lo}-{hi}.{self.rest_format}")
+            else:
+                clause = f"{attribute}>{lo} AND {attribute}<={hi}"
+                get_dict["where"] = f"{clause} AND ({base_where})" if base_where else clause
+                url = f"{layer_url}?{urlencode(get_dict)}"
+                if cache_fs is not None:
+                    cache_file = os.path.join(
+                        cache_folder, f"{prefix}{lo}-{hi}.{self.rest_format}"
+                    )
+                    try:
+                        with cache_fs.open(cache_file, mode="wb") as file:
+                            stream_file(source_fs, url, file)
+                    except Exception:
+                        # A download that broke off must not survive as a cached page
+                        if cache_fs.exists(cache_file):
+                            cache_fs.rm(cache_file)
+                        raise
+                    url = cache_file
+
+            try:
+                data = gpd.read_file(url)
+            except Exception as e:
+                # An error response from the server must not survive as a cached page
+                if cache_fs is not None and cache_fs.exists(url):
+                    cache_fs.rm(url)
+                raise RuntimeError(
+                    f"Could not read ids ({lo} ... {hi}] of {layer_url}: {e}"
+                ) from e
+
+            if len(data) == 0 and not cached:
+                # An id gap wider than a page: ask once where the ids resume, and
+                # let the empty page cover the whole gap on later runs, instead of
+                # paging through a span that may hold millions of absent ids.
+                resume = self._rest_id_bound(layer_url, attribute, base_where, "ASC", floor=lo)
+                if resume - 1 > hi and cache_fs is not None:
+                    gap = os.path.join(cache_folder, f"{prefix}{lo}-{resume - 1}.{self.rest_format}")
+                    cache_fs.mv(url, gap)
+                hi = max(hi, resume - 1)
 
-            yield data, base_url, base_url, layer["id"]
+            lo = hi
+            if len(data) == 0:
+                continue
+            self.info(f"Read {len(data)} features, page {page} up to id {hi}")
+            page += 1
+            yield self._unqualify(data), base_url, base_url, layer["id"]
 
-            if not len(data) >= page_size:
-                break
+    @staticmethod
+    def _cached_windows(cache_fs, cache_folder, prefix):
+        """The cached (lo, hi] windows, keyed by lo. The name carries both bounds,
+        so a page is only ever read as exactly the window it was fetched for."""
+        if cache_fs is None or not cache_fs.exists(cache_folder):
+            return {}
+        pattern = re.compile(re.escape(prefix) + r"(-?\d+)-(-?\d+)\.")
+        windows = {}
+        for path in cache_fs.ls(cache_folder, detail=False):
+            match = pattern.search(os.path.basename(str(path)))
+            if match:
+                windows[int(match.group(1))] = int(match.group(2))
+        return windows
+
+    def _rest_id_bound(self, layer_url, attribute, base_where, direction, floor=-1, attempts=5):
+        clause = f"{attribute}>{floor}"
+        params = {
+            "f": "json",
+            "where": f"{clause} AND ({base_where})" if base_where else clause,
+            "outFields": attribute,
+            "returnGeometry": "false",
+            "orderByFields": f"{attribute} {direction}",
+            "resultRecordCount": 1,
+        }
+        # This is the one sorted query left, and it is the one a tired server
+        # gives up on: the Balearic proxy answers two in three with a 502. The
+        # pages themselves are range queries and do not need this.
+        for attempt in range(attempts):
+            try:
+                response = requests.get(layer_url, params).json()
+                return int(next(iter(response["features"][0]["attributes"].values())))
+            except Exception:
+                if attempt == attempts - 1:
+                    raise
+                time.sleep(2**attempt)

fiboa_cli/datasets/es_cb.py

@@ -9,14 +9,16 @@ class ESCBConverter(EsriRESTConverterMixin, ESBaseConverter):
     short_name = "Spain Cantabria"
     title = "Spain Cantabria Crop fields"
     description = "SIGPAC Crop fields of Spain - Cantabria"
-    # https://www.caib.es/sites/M170613081930629/f/463418
-    # see https://intranet.caib.es/opendatacataleg/dataset/sigpac-2024/resource/3a0bc2e0-3f37-45b7-a7d4-1e8c7cf09bc8
-    # "Our licenses allow the reproduction or redistribution of the licensed digital information to third parties. In such cases, it is essential that when redistributing or transferring the data to said third parties, they clearly and explicitly accept the conditions of our non-commercial use license."
-    license = "CC-BY-NC-4.0"  # http://www.opendefinition.org/licenses/cc-by
+    # Not Creative Commons: Decreto 87/2013 (modified by 102/2018) defines two
+    # licences of its own, both free of charge, the commercial one needed only
+    # for reselling. The service names no licence, only the copyright holder.
+    # https://www.territoriodecantabria.es/cartografia-sig/descargas-y-politica-de-licencias/preguntas-frecuentes
+    license = "Licencia de uso de datos del Gobierno de Cantabria (Decreto 87/2013) <https://www.territoriodecantabria.es/cartografia-sig/datos-abiertos-y-politica-de-licencias>"
+    # The wording the licence requires for original data, verbatim.
     attribution = (
-        "©Government of Cantabria. Free information available at https://mapas.cantabria.es"
+        "© Gobierno de Cantabria. Información gratuita disponible en https://mapas.cantabria.es"
     )
-    provider = ""
+    provider = "Gobierno de Cantabria <https://mapas.cantabria.es>"
     columns = {
         "DN_OID": "id",
         "geometry": "geometry",
@@ -35,16 +37,20 @@ class ESCBConverter(EsriRESTConverterMixin, ESBaseConverter):
         }
     }
 
-    variants = {str(year): str(year) for year in range(2024, 2010 - 1, -1)}
+    variants = {str(year): str(year) for year in range(2025, 2010 - 1, -1)}
     use_code_attribute = "USO_SIGPAC"
+    use_variant_as_determination = True
 
     # "https://geoservicios.cantabria.es/inspire/rest/services/SIGPAC/MapServer?f=json"
     # "https://geoservicios.cantabria.es/inspire/rest/services/SIGPAC/MapServer/63/query?f=json&where=1%3D1&spatialRel=esriSpatialRelIntersects&geometry=%7B%22xmin%22%3A407913.2828037373%2C%22ymin%22%3A4804384.359524686%2C%22xmax%22%3A411054.4224193499%2C%22ymax%22%3A4805366.49482229%2C%22spatialReference%22%3A%7B%22wkid%22%3A25830%2C%22latestWkid%22%3A25830%7D%7D&geometryType=esriGeometryEnvelope&inSR=25830&outFields=OBJECTID%2CPROVINCIA%2CMUNICIPIO%2CAGREGADO%2CZONA%2CPOLIGONO%2CPARCELA%2CRECINTO%2CUSO_SIGPAC%2CSHAPE_Area&orderByFields=OBJECTID%20ASC&outSR=25830"
 
+    # 2010-2014 are joined layers, whose fields arrive table-qualified
+    # (SIGPAC_2014_ATRRE.USO_SIGPAC); the REST mixin strips the prefixes.
     rest_base_url = "https://geoservicios.cantabria.es/inspire/rest/services/SIGPAC/MapServer"
     # rest_params = {"where": "USO_SIGPAC NOT IN ('AG','CA','ED','FO','IM','IS','IV','TH','ZC','ZU','ZV','MT')"}
 
     def rest_layer_filter(self, layers):
-        self.column_additions["determination:datetime"] = ""
+        if not self.variant:
+            self.variant = next(iter(self.variants))
         regex = re.compile("Recintos SIGPAC " + self.variant)
         return next(layer for layer in layers if regex.match(layer["name"]))