fix: fetch every resource for the deleted entities expectation

digital_land/expectations/operations/dataset.py

@@ -135,35 +135,58 @@ def count_lpa_boundary(
     return result, message, details
 
 
+# datasette's max_returned_rows, a query returning more rows than this is cut off without an error
+DATASETTE_PAGE_SIZE = 1000
+
+
 def _fetch_resources_for_dataset(dataset_name, active_only):
     if active_only:
-        # one active resource per endpoint
+        # one active resource per endpoint, ignoring log rows for days the endpoint returned no resource
         sql = f"""select o.entity as organisation_entity, rhe.resource
                   from reporting_historic_endpoints rhe
                   join organisation o on rhe.organisation=o.organisation
-                  where pipeline == '{dataset_name}' and (resource_end_date == "" or resource_end_date is null)
-                  group by o.entity, rhe.endpoint"""
+                  where pipeline == '{dataset_name}' and rhe.resource != ""
+                  and (resource_end_date == "" or resource_end_date is null)
+                  group by o.entity, rhe.endpoint
+                  order by o.entity, rhe.endpoint"""
     else:
         # every resource each organisation has ever supplied, including ended ones
         sql = f"""select o.entity as organisation_entity, rhe.resource
                   from reporting_historic_endpoints rhe
                   join organisation o on rhe.organisation=o.organisation
                   where pipeline == '{dataset_name}' and rhe.resource != ""
-                  group by o.entity, rhe.resource"""
+                  group by o.entity, rhe.resource
+                  order by o.entity, rhe.resource"""
+
+    pages = []
+    offset = 0
+    while True:
+        page = _read_datasette_csv(
+            f"{sql} limit {DATASETTE_PAGE_SIZE} offset {offset}", dataset_name
+        )
+        pages.append(page)
+        if len(page) < DATASETTE_PAGE_SIZE:
+            break
+        offset += DATASETTE_PAGE_SIZE
+    df = pd.concat(pages, ignore_index=True)
+
+    logging.warning(
+        f"[expectations] Fetched {len(df)} {'active' if active_only else 'historic'} resources for '{dataset_name}' from datasette"
+    )
+    cache = {}
+    for _, row in df.iterrows():
+        cache.setdefault(row["organisation_entity"], []).append(row["resource"])
+    return cache
+
+
+def _read_datasette_csv(sql, dataset_name):
     params = urllib.parse.urlencode({"sql": sql, "_size": "max"})
     base_url = f"https://datasette.planning.data.gov.uk/digital-land.csv?{params}"
 
     max_retries = 60
     for attempt in range(max_retries):
         try:
-            df = pd.read_csv(base_url)
-            logging.warning(
-                f"[expectations] Fetched {len(df)} {'active' if active_only else 'historic'} resources for '{dataset_name}' from datasette"
-            )
-            cache = {}
-            for _, row in df.iterrows():
-                cache.setdefault(row["organisation_entity"], []).append(row["resource"])
-            return cache
+            return pd.read_csv(base_url)
         except urllib.error.HTTPError as e:
             logging.warning(
                 f"HTTP error fetching datasette for dataset {dataset_name}, "

tests/integration/expectations/operations/test_dataset.py

@@ -4,6 +4,7 @@
 import pytest
 import pandas as pd
 import requests
+import urllib
 
 from digital_land.expectations.operations.dataset import (
     check_columns,
@@ -154,6 +155,69 @@ def test_fetch_all_resources_for_dataset_includes_ended_resources(mocker):
     assert "resource_end_date" not in mock_read_csv.call_args.args[0]
 
 
+@pytest.fixture
+def fake_datasette(mocker):
+    """Answer datasette CSV queries from an in-memory reporting database, cut off at
+    datasette's max_returned_rows like the real service."""
+    conn = sqlite3.connect(":memory:")
+    conn.execute("CREATE TABLE organisation (organisation TEXT, entity INTEGER)")
+    conn.execute(
+        """CREATE TABLE reporting_historic_endpoints (
+            organisation TEXT, pipeline TEXT, endpoint TEXT, resource TEXT, resource_end_date TEXT
+        )"""
+    )
+
+    def read_csv(url):
+        sql = urllib.parse.parse_qs(urllib.parse.urlparse(url).query)["sql"][0]
+        return pd.read_sql_query(sql, conn).head(1000)
+
+    mocker.patch("pandas.read_csv", side_effect=read_csv)
+    return conn
+
+
+def test_fetch_active_resources_for_dataset_ignores_rows_without_a_resource(
+    fake_datasette,
+):
+    fake_datasette.execute(
+        "INSERT INTO organisation VALUES ('local-authority:LBH', 192)"
+    )
+    # log rows for days the endpoint returned no resource, then its active resource
+    fake_datasette.executemany(
+        "INSERT INTO reporting_historic_endpoints VALUES (?, ?, ?, ?, ?)",
+        [("local-authority:LBH", "test-dataset", "endpoint_a", "", None)] * 3
+        + [("local-authority:LBH", "test-dataset", "endpoint_a", "resource_a", "")],
+    )
+
+    result = fetch_active_resources_for_dataset("test-dataset")
+
+    assert result == {192: ["resource_a"]}
+
+
+def test_fetch_all_resources_for_dataset_reads_past_the_datasette_row_limit(
+    fake_datasette,
+):
+    fake_datasette.execute(
+        "INSERT INTO organisation VALUES ('local-authority:LBH', 192)"
+    )
+    fake_datasette.executemany(
+        "INSERT INTO reporting_historic_endpoints VALUES (?, ?, ?, ?, ?)",
+        [
+            (
+                "local-authority:LBH",
+                "test-dataset",
+                "endpoint_a",
+                f"resource_{i:04}",
+                "2026-01-01",
+            )
+            for i in range(2500)
+        ],
+    )
+
+    result = fetch_all_resources_for_dataset("test-dataset")
+
+    assert len(result[192]) == 2500
+
+
 def test_count_deleted_entities(dataset_path, mocker):
     """Entity with no facts in the active resource is reported as deleted."""
     organisation_entity = 109