fix: use WFS POST for historical land use

scripts/provision_mol_historical_landuse.py

@@ -78,7 +78,7 @@ def build_session() -> requests.Session:
         status=5,
         backoff_factor=1.0,
         status_forcelist=(429, 500, 502, 503, 504),
-        allowed_methods=frozenset({"GET"}),
+        allowed_methods=frozenset({"GET", "POST"}),
         raise_on_status=True,
     )
     session = requests.Session()
@@ -127,6 +127,54 @@ def wfs_filter_xml(definition: ThemeDefinition, bounds: tuple[float, float, floa
     )
 
 
+def wfs_request_xml(
+    *,
+    collection: str,
+    filter_xml: str,
+    page_size: int,
+    start_index: int,
+) -> str:
+    return (
+        "<?xml version='1.0' encoding='UTF-8'?>"
+        "<wfs:GetFeature service='WFS' version='2.0.0' outputFormat='application/json' "
+        f"count='{page_size}' startIndex='{start_index}' "
+        "xmlns:wfs='http://www.opengis.net/wfs/2.0' "
+        "xmlns:fes='http://www.opengis.net/fes/2.0' "
+        "xmlns:gml='http://www.opengis.net/gml/3.2' "
+        "xmlns:HistLandgebruik='https://geo.api.vlaanderen.be/HistLandgebruik'>"
+        f"<wfs:Query typeNames='{collection}' srsName='EPSG:4326'>{filter_xml}</wfs:Query>"
+        "</wfs:GetFeature>"
+    )
+
+
+def fetch_wfs_page(
+    session: requests.Session,
+    *,
+    collection: str,
+    filter_xml: str,
+    page_size: int,
+    start_index: int,
+    timeout: int,
+) -> list[dict[str, Any]]:
+    response = session.post(
+        WFS_URL,
+        data=wfs_request_xml(
+            collection=collection,
+            filter_xml=filter_xml,
+            page_size=page_size,
+            start_index=start_index,
+        ).encode("utf-8"),
+        headers={"Content-Type": "application/xml; charset=UTF-8", "Accept": "application/json"},
+        timeout=timeout,
+    )
+    response.raise_for_status()
+    payload = response.json()
+    features = payload.get("features") if isinstance(payload, dict) else None
+    if not isinstance(features, list):
+        raise RuntimeError("Historical land-use WFS returned an invalid FeatureCollection")
+    return features
+
+
 def fetch_year(
     session: requests.Session,
     year: int,
@@ -145,23 +193,14 @@ def fetch_year(
         start_index = 0
         seen: set[str] = set()
         while True:
-            response = session.get(
-                WFS_URL,
-                params={
-                    "service": "WFS",
-                    "version": "2.0.0",
-                    "request": "GetFeature",
-                    "typeNames": collection,
-                    "outputFormat": "application/json",
-                    "srsName": "EPSG:4326",
-                    "FILTER": filter_xml,
-                    "count": page_size,
-                    "startIndex": start_index,
-                },
+            page = fetch_wfs_page(
+                session,
+                collection=collection,
+                filter_xml=filter_xml,
+                page_size=page_size,
+                start_index=start_index,
                 timeout=timeout,
             )
-            response.raise_for_status()
-            page = response.json().get("features") or []
             for raw_feature in page:
                 feature_id = str(raw_feature.get("id") or "")
                 if not feature_id or feature_id in seen:
@@ -234,23 +273,14 @@ def write_theme_snapshot(
             )
             output.write(',"features":[')
             while True:
-                response = session.get(
-                    WFS_URL,
-                    params={
-                        "service": "WFS",
-                        "version": "2.0.0",
-                        "request": "GetFeature",
-                        "typeNames": collection,
-                        "outputFormat": "application/json",
-                        "srsName": "EPSG:4326",
-                        "FILTER": filter_xml,
-                        "count": page_size,
-                        "startIndex": start_index,
-                    },
+                page = fetch_wfs_page(
+                    session,
+                    collection=collection,
+                    filter_xml=filter_xml,
+                    page_size=page_size,
+                    start_index=start_index,
                     timeout=timeout,
                 )
-                response.raise_for_status()
-                page = response.json().get("features") or []
                 for raw_feature in page:
                     feature_id = str(raw_feature.get("id") or "")
                     if not feature_id or feature_id in seen: