feat: validate datastore write records

datastore/services/write.py

@@ -1,9 +1,14 @@
 from __future__ import annotations
 
 import asyncio
+import json
 from datetime import datetime, timezone
 from typing import TYPE_CHECKING, Any
 
+from frictionless import Resource, Schema
+from frictionless.exception import FrictionlessException
+
+from datastore.core.exceptions import ValidationError
 from datastore.infrastructure.engines import get_datastore_engine
 from datastore.schemas.responses import (
     DatastoreCreateResponse,
@@ -21,6 +26,50 @@ def _utc_now_iso() -> str:
     return datetime.now(timezone.utc).replace(tzinfo=None).isoformat()
 
 
+def _validate_records(
+    resource_id: str,
+    schema: dict[str, Any],
+    records: list[dict[str, Any]],
+) -> None:
+    """Validate in-memory records against a stored Frictionless schema.
+
+    This deliberately lives in the service layer so every storage backend gets
+    the same row validation behavior. Backends only provide the canonical
+    schema through `InfoResult` and remain responsible for persistence.
+    """
+    if not records or not schema.get("fields"):
+        return
+
+    try:
+        schema_obj = Schema.from_descriptor(schema)
+        resource = Resource(data=records, schema=schema_obj, format="inline")
+        report = resource.validate(
+            parallel=False,
+            limit_errors=1000,
+        )
+    except FrictionlessException as exc:
+        raise ValidationError(
+            f"Stored schema for resource {resource_id!r} failed validation: {exc}"
+        ) from exc
+
+    if report.valid:
+        return
+
+    report_dict = report.to_dict()
+    messages: list[str] = []
+    for task in report_dict.get("tasks", []):
+        for error in task.get("errors", []):
+            message = error.get("message") or "Invalid record"
+            messages.append(str(message))
+
+    error_count = report.stats.get("errors", len(messages))
+    raise ValidationError(
+        f"Records failed Frictionless validation for resource {resource_id!r}: "
+        f"{error_count} error(s)",
+        fields={"records": messages or [json.dumps(report_dict)]},
+    )
+
+
 async def _sync_resource_to_ckan(
     context: RequestContext,
     resource_id: str,
@@ -66,6 +115,7 @@ async def create_datastore(
     include_total = bool(data_dict.get("include_total", False))
 
     fields, primary_key = frictionless_schema_to_fields(schema)
+    _validate_records(resource_id=str(resource or "new-resource"), schema=schema, records=records)
 
     if isinstance(resource, dict):
         # Endpoint gates this branch on AUTH_TYPE=ckan, so context.ckan is
@@ -122,6 +172,9 @@ async def upsert_datastore(
     include_total = bool(data_dict.get("include_total", False))
 
     engine = get_datastore_engine(context, mode="rw")
+    info = await asyncio.to_thread(engine.info, resource_id)
+    _validate_records(resource_id=resource_id, schema=info.schema, records=records)
+
     write_result = await asyncio.to_thread(
         engine.upsert,
         resource_id=resource_id,

tests/test_write_service.py

@@ -13,6 +13,8 @@
 
 import pytest
 from datastore.core.config import Config
+from datastore.core.exceptions import ValidationError
+from datastore.infrastructure.engines.base import InfoResult
 from datastore.services.write import (
     create_datastore,
     delete_datastore,
@@ -77,6 +79,29 @@ def test_existing_resource_skips_resource_create() -> None:
     assert ctx.ckan.created == []  # no CKAN call
 
 
+def test_create_rejects_invalid_records_before_creating_resource() -> None:
+    ctx = _ctx()
+    data_dict = {
+        "package": {"id": "pkg-1"},
+        "resource": {"package_id": "pkg-1", "name": "foo"},
+        "schema": {
+            "fields": [
+                {
+                    "name": "status",
+                    "type": "string",
+                    "constraints": {"enum": ["active", "inactive"]},
+                }
+            ]
+        },
+        "records": [{"status": "pending"}],
+    }
+
+    with pytest.raises(ValidationError, match="Frictionless validation"):
+        asyncio.run(create_datastore(ctx, data_dict))
+
+    assert ctx.ckan.created == []
+
+
 def test_new_resource_creates_via_ckan() -> None:
     ctx = _ctx()
     data_dict = {
@@ -288,6 +313,50 @@ def test_upsert_returns_typed_result() -> None:
     assert result.total is None
 
 
+def test_upsert_rejects_invalid_records_before_engine_write(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    from datastore.infrastructure.engines.bigquery import BigQueryBackend
+
+    schema = {
+        "fields": [
+            {
+                "name": "amount",
+                "type": "number",
+                "constraints": {"minimum": 0, "maximum": 100},
+            }
+        ]
+    }
+    writes: list[list[dict[str, Any]]] = []
+
+    monkeypatch.setattr(
+        BigQueryBackend,
+        "info",
+        lambda self, resource_id: InfoResult(schema=schema, meta={}),
+    )
+
+    def fake_upsert(self: Any, **kwargs: Any) -> dict[str, Any]:
+        writes.append(kwargs["records"])
+        return {"total": 1}
+
+    monkeypatch.setattr(BigQueryBackend, "upsert", fake_upsert)
+
+    with pytest.raises(ValidationError) as exc_info:
+        asyncio.run(
+            upsert_datastore(
+                _ctx(),
+                {
+                    "resource_id": "res-1",
+                    "records": [{"amount": 101}],
+                    "method": "upsert",
+                },
+            )
+        )
+
+    assert "maximum" in exc_info.value.fields["records"][0]
+    assert writes == []
+
+
 def test_upsert_default_method_is_upsert() -> None:
     """`method` is optional; absence resolves to 'upsert' inside the service."""
     ctx = _ctx()