From d38b32a4e378ae55ea433772639bea94366d2005 Mon Sep 17 00:00:00 2001 From: Luccas Gomes Date: Fri, 7 Aug 2026 16:04:12 -0300 Subject: [PATCH] feat: validate datastore write records --- datastore/services/write.py | 53 ++++++++++++++++++++++++++++ tests/test_write_service.py | 69 +++++++++++++++++++++++++++++++++++++ 2 files changed, 122 insertions(+) diff --git a/datastore/services/write.py b/datastore/services/write.py index 6a7ce7a..dfcccb3 100644 --- a/datastore/services/write.py +++ b/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, diff --git a/tests/test_write_service.py b/tests/test_write_service.py index c0c2dcb..c130766 100644 --- a/tests/test_write_service.py +++ b/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()