diff --git a/datastore/api/endpoints/datastore.py b/datastore/api/endpoints/datastore.py index 32a8cf0..6fb6618 100644 --- a/datastore/api/endpoints/datastore.py +++ b/datastore/api/endpoints/datastore.py @@ -90,6 +90,9 @@ async def datastore_create( data_dict.update( { + # Captured before `resource` is overwritten: the CKAN sync only + # touches `url` / `last_modified` on datastore-owned resources. + "url_type": data_dict["resource"].get("url_type"), "resource": payload.resource_id or payload.resource, "schema": payload.schema, "records": payload.records, diff --git a/datastore/services/write.py b/datastore/services/write.py index bde2ee2..58b4de5 100644 --- a/datastore/services/write.py +++ b/datastore/services/write.py @@ -95,9 +95,13 @@ async def _sync_resource_to_ckan( ) -> None: """Keep the CKAN resource metadata in sync with the datastore table. - Always refreshes `last_modified` — and, because `resource_patch` runs - CKAN's `resource_update`, records an entry in the resource's activity - log. The schema is touched per the operation: + `resource_patch` runs CKAN's `resource_update`, so every sync records + an entry in the resource's activity log. `last_modified` and `url` are + only written when the datastore owns the resource + (`url_type="datastore"`). On an upload or link they describe the + source file, and CKAN flags a change to either as `url_changed`, which + makes ckanext-aircan re-submit the resource to the ingestion pipeline. + The schema is touched per the operation: - `schema=` — an op that (re)defines columns (`datastore_create`, or a `datastore_delete` column drop): mirror it so CKAN matches the @@ -114,7 +118,10 @@ async def _sync_resource_to_ckan( """ if context.ckan is None or not resource_id: return - patch: dict[str, Any] = {"last_modified": _utc_now_iso()} + owned = url_type == "datastore" + patch: dict[str, Any] = {} + if owned: + patch["last_modified"] = _utc_now_iso() if table_dropped: patch["datastore_active"] = False if url_type != "upload": @@ -122,9 +129,10 @@ async def _sync_resource_to_ckan( patch["url"] = None elif schema is not None: patch["schema"] = schema - if dump_url is not None: + if dump_url is not None and owned: patch["url"] = dump_url - await context.ckan.resource_patch(resource_id=resource_id, patch=patch) + if patch: + await context.ckan.resource_patch(resource_id=resource_id, patch=patch) async def create_datastore( @@ -157,8 +165,10 @@ async def create_datastore( } ) resource_id = resource["id"] + url_type = "datastore" else: resource_id = resource + url_type = data_dict.get("url_type") engine = get_datastore_engine(context, mode="rw") # Off the event loop — BigQuery's sync client would otherwise block @@ -175,6 +185,7 @@ async def create_datastore( context, resource_id, schema=schema, + url_type=url_type, dump_url=_dump_url(context.config, resource_id), ) @@ -211,7 +222,8 @@ async def upsert_datastore( include_total=include_total, ) - await _sync_resource_to_ckan(context, resource_id) + resource = data_dict.get("resource") or {} + await _sync_resource_to_ckan(context, resource_id, url_type=resource.get("url_type")) return DatastoreUpsertResponse.Result( resource_id=resource_id, @@ -242,19 +254,14 @@ async def delete_datastore( # Sync CKAN per the delete variant (mirrors the engine's branching): # column drop → mirror the reduced schema # whole-table drop (no filters, no fields) → drop the schema too - # row delete → data only (timestamp + activity) + # row delete → data only (timestamp + activity, if owned) + url_type = (data_dict.get("resource") or {}).get("url_type") if fields is not None: - await _sync_resource_to_ckan(context, resource_id, schema=result.schema) + await _sync_resource_to_ckan(context, resource_id, schema=result.schema, url_type=url_type) elif filters is None: - resource = data_dict.get("resource") or {} - await _sync_resource_to_ckan( - context, - resource_id, - table_dropped=True, - url_type=resource.get("url_type"), - ) + await _sync_resource_to_ckan(context, resource_id, table_dropped=True, url_type=url_type) else: - await _sync_resource_to_ckan(context, resource_id) + await _sync_resource_to_ckan(context, resource_id, url_type=url_type) return DatastoreDeleteResponse.Result( resource_id=resource_id, diff --git a/tests/test_datastore_create.py b/tests/test_datastore_create.py index f94952f..96d7f5b 100644 --- a/tests/test_datastore_create.py +++ b/tests/test_datastore_create.py @@ -446,3 +446,31 @@ def test_create_on_readonly_resource_with_force_succeeds( assert response.status_code == 200 assert response.json()["success"] is True + + +def test_forced_create_on_upload_keeps_url_and_last_modified( + client: TestClient, fake_ckan: FakeCKAN +) -> None: + """Updating the data dictionary of an uploaded resource must not rewrite + its file `url` or `last_modified` — CKAN reads either as a new upload and + ckanext-aircan would re-run the ingestion pipeline.""" + fake_ckan.add_resource( + "ro-res", + package_id="pkg-balancing-2025", + url_type="upload", + url="flows.csv", + last_modified="2026-01-01T00:00:00", + ) + payload = { + "resource_id": "ro-res", + "force": True, + "fields": [{"id": "auction_id", "type": "int4", "info": {"title": "Auction"}}], + } + + response = client.post(CREATE_URL, json=payload) + + assert response.status_code == 200 + resource = fake_ckan.resources["ro-res"] + assert resource["url"] == "flows.csv" + assert resource["last_modified"] == "2026-01-01T00:00:00" + assert resource["schema"]["fields"][0]["title"] == "Auction" diff --git a/tests/test_write_service.py b/tests/test_write_service.py index f41f34d..85ec737 100644 --- a/tests/test_write_service.py +++ b/tests/test_write_service.py @@ -126,6 +126,7 @@ def test_create_syncs_schema_to_ckan_resource() -> None: data_dict = { "package": {"id": "pkg-1"}, "resource": "res-1", + "url_type": "datastore", "schema": schema, "records": [], } @@ -137,6 +138,29 @@ def test_create_syncs_schema_to_ckan_resource() -> None: assert rid == "res-1" assert patch["schema"] == schema assert isinstance(patch["last_modified"], str) and patch["last_modified"] + assert patch["url"].endswith("/res-1") + + +@pytest.mark.parametrize("url_type", ["upload", "", None]) +def test_create_on_non_datastore_resource_syncs_schema_only(url_type: str | None) -> None: + """A forced create on an upload / link resource must not touch `url` or + `last_modified`: they describe the source file, and CKAN treats a change + to either as a new upload, which re-runs the ingestion pipeline.""" + ctx = _ctx() + schema = _schema(primary_key=["a"]) + data_dict = { + "package": {"id": "pkg-1"}, + "resource": "res-1", + "url_type": url_type, + "schema": schema, + "records": [], + } + + asyncio.run(create_datastore(ctx, data_dict)) + + rid, patch = ctx.ckan.patched[0] + assert rid == "res-1" + assert patch == {"schema": schema} def test_create_new_resource_also_syncs_schema() -> None: @@ -156,6 +180,8 @@ def test_create_new_resource_also_syncs_schema() -> None: rid, patch = ctx.ckan.patched[0] assert rid == "new-res-id" # id returned by resource_create assert patch["schema"] == schema + assert patch["last_modified"] # the datastore owns a resource it created + assert patch["url"].endswith("/new-res-id") def test_create_without_ckan_skips_schema_sync() -> None: @@ -178,7 +204,12 @@ def test_upsert_syncs_timestamp_to_ckan() -> None: """Upsert changes data, not columns — it refreshes the resource timestamp (and activity log) but must NOT touch the schema.""" ctx = _ctx() - data_dict = {"resource_id": "res-1", "records": [{"a": 1}], "method": "upsert"} + data_dict = { + "resource_id": "res-1", + "resource": {"url_type": "datastore"}, + "records": [{"a": 1}], + "method": "upsert", + } asyncio.run(upsert_datastore(ctx, data_dict)) @@ -190,11 +221,31 @@ def test_upsert_syncs_timestamp_to_ckan() -> None: assert "url_type" not in patch +def test_upsert_on_non_datastore_resource_skips_ckan_sync() -> None: + """A forced upsert into an upload has nothing safe to sync: the schema + is unchanged and `last_modified` would re-trigger ingestion.""" + ctx = _ctx() + data_dict = { + "resource_id": "res-1", + "resource": {"url_type": "upload"}, + "records": [{"a": 1}], + "method": "upsert", + } + + asyncio.run(upsert_datastore(ctx, data_dict)) + + assert ctx.ckan.patched == [] + + def test_delete_rows_syncs_timestamp_only() -> None: """Row delete (placeholder yields no schema) → timestamp + activity, no schema patch.""" ctx = _ctx() - data_dict = {"resource_id": "res-1", "filters": {"a": 1}} + data_dict = { + "resource_id": "res-1", + "resource": {"url_type": "datastore"}, + "filters": {"a": 1}, + } asyncio.run(delete_datastore(ctx, data_dict)) @@ -219,7 +270,11 @@ def fake_delete(self: Any, *, resource_id: str, filters: Any, fields: Any) -> An monkeypatch.setattr(BigQueryBackend, "delete", fake_delete) ctx = _ctx() - data_dict = {"resource_id": "res-1", "fields": ["b"]} + data_dict = { + "resource_id": "res-1", + "resource": {"url_type": "datastore"}, + "fields": ["b"], + } asyncio.run(delete_datastore(ctx, data_dict)) @@ -234,7 +289,8 @@ def test_delete_whole_table_clears_schema_on_ckan() -> None: BigQuery table — so the resource's schema is dropped too, keeping the two consistent.""" ctx = _ctx() - data_dict = {"resource_id": "res-1"} # no filters, no fields → drop table + # no filters, no fields → drop table + data_dict = {"resource_id": "res-1", "resource": {"url_type": "datastore"}} asyncio.run(delete_datastore(ctx, data_dict))