diff --git a/digital_land/expectations/operations/dataset.py b/digital_land/expectations/operations/dataset.py index 502bdfac..43e8e6dd 100644 --- a/digital_land/expectations/operations/dataset.py +++ b/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}, " diff --git a/tests/integration/expectations/operations/test_dataset.py b/tests/integration/expectations/operations/test_dataset.py index c951683a..21a00387 100644 --- a/tests/integration/expectations/operations/test_dataset.py +++ b/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