fix(sync): don't touch url/last_modified on non-datastore resources

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,

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=<dict>` — an op that (re)defines columns (`datastore_create`,
       or a `datastore_delete` column drop): mirror it so CKAN matches the
@@ -114,17 +118,21 @@ 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":
             patch["schema"] = None
             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,

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"

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))