From 1974bf1a2b27ad97b724294c39d0b20071d73f79 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Tue, 4 Aug 2026 12:05:09 +0200 Subject: [PATCH] Define bounded tabular source contracts --- docs/TABULAR_SOURCE_CONTRACT.md | 21 ++++++++ src/govoplan_core/core/datasources.py | 27 ++++++++++ src/govoplan_core/core/tabular_sources.py | 56 ++++++++++++++++++- tests/test_tabular_source_contract.py | 66 +++++++++++++++++++++++ 4 files changed, 169 insertions(+), 1 deletion(-) create mode 100644 docs/TABULAR_SOURCE_CONTRACT.md diff --git a/docs/TABULAR_SOURCE_CONTRACT.md b/docs/TABULAR_SOURCE_CONTRACT.md new file mode 100644 index 0000000..c516f35 --- /dev/null +++ b/docs/TABULAR_SOURCE_CONTRACT.md @@ -0,0 +1,21 @@ +# Tabular Source Preview Contract + +Core defines provider-neutral DTOs for optional tabular source providers. A +source declares whether it is live, cached, file-backed, or static; its schema +and immutable fingerprint; structured health; and the exact projection, +pagination, filter, aggregation, and sorting operations that the provider can +push down. Consumers must not infer pushdown support from a provider name. + +Every preview request carries independent row, byte, and elapsed-time budgets. +A provider may tighten these values but must return its effective limits, +returned byte count, elapsed milliseconds, truncation state, and structured +diagnostics. Equivalent fields on the Datasources read request and result +preserve that evidence when a live source is consumed through the catalogue. +A row that cannot fit within the byte budget fails explicitly rather than +leaking a partial value. Timeout, stale fingerprint, unavailable source, and +authorization failures remain distinct provider-neutral errors. + +Connector health and preview diagnostics must contain no credentials, endpoint +userinfo, row values, or unbounded remote error bodies. A Datasource origin +preserves this contract so registration and staging do not erase source mode, +health, pushdown, or preview-limit evidence. diff --git a/src/govoplan_core/core/datasources.py b/src/govoplan_core/core/datasources.py index de76f72..c514826 100644 --- a/src/govoplan_core/core/datasources.py +++ b/src/govoplan_core/core/datasources.py @@ -9,6 +9,14 @@ from govoplan_core.core.external_references import ( SOURCE_AUTHORITY_MODES, SourceAuthorityMode, ) +from govoplan_core.core.tabular_sources import ( + DEFAULT_PREVIEW_BYTES, + DEFAULT_PREVIEW_TIMEOUT_MS, + TabularPreviewDiagnostic, + TabularPushdown, + TabularSourceHealth, + TabularSourceMode, +) CAPABILITY_DATASOURCE_CATALOGUE = "datasources.catalogue" @@ -265,6 +273,8 @@ class DatasourceReadRequest: offset: int = 0 columns: tuple[str, ...] = () expected_fingerprint: str | None = None + max_bytes: int = DEFAULT_PREVIEW_BYTES + timeout_ms: int = DEFAULT_PREVIEW_TIMEOUT_MS @dataclass(frozen=True, slots=True) @@ -274,6 +284,12 @@ class DatasourceReadResult: total_rows: int truncated: bool materialization: DatasourceMaterialization | None = None + returned_bytes: int = 0 + elapsed_ms: int = 0 + effective_row_limit: int = 0 + effective_byte_limit: int = 0 + effective_timeout_ms: int = 0 + diagnostics: tuple[TabularPreviewDiagnostic, ...] = () @dataclass(frozen=True, slots=True) @@ -339,6 +355,9 @@ class DatasourceOrigin: updated_at: datetime | None = None capabilities: tuple[str, ...] = ("read",) metadata: Mapping[str, object] = field(default_factory=dict) + source_mode: TabularSourceMode = "cached" + pushdown: TabularPushdown = field(default_factory=TabularPushdown) + health: TabularSourceHealth = field(default_factory=TabularSourceHealth) @dataclass(frozen=True, slots=True) @@ -348,6 +367,8 @@ class DatasourceOriginReadRequest: offset: int = 0 columns: tuple[str, ...] = () expected_fingerprint: str | None = None + max_bytes: int = DEFAULT_PREVIEW_BYTES + timeout_ms: int = DEFAULT_PREVIEW_TIMEOUT_MS @dataclass(frozen=True, slots=True) @@ -356,6 +377,12 @@ class DatasourceOriginReadResult: rows: tuple[Mapping[str, object], ...] total_rows: int truncated: bool + returned_bytes: int = 0 + elapsed_ms: int = 0 + effective_row_limit: int = 0 + effective_byte_limit: int = 0 + effective_timeout_ms: int = 0 + diagnostics: tuple[TabularPreviewDiagnostic, ...] = () @runtime_checkable diff --git a/src/govoplan_core/core/tabular_sources.py b/src/govoplan_core/core/tabular_sources.py index ccfd0b5..680832c 100644 --- a/src/govoplan_core/core/tabular_sources.py +++ b/src/govoplan_core/core/tabular_sources.py @@ -6,11 +6,17 @@ import re from collections.abc import Mapping, Sequence from dataclasses import dataclass, field from datetime import datetime -from typing import Protocol, runtime_checkable +from typing import Literal, Protocol, runtime_checkable CAPABILITY_CONNECTORS_TABULAR_SOURCES = "connectors.tabularSources" CAPABILITY_CONNECTORS_TABULAR_SNAPSHOT_WRITER = "connectors.tabularSnapshotWriter" +DEFAULT_PREVIEW_BYTES = 1_000_000 +DEFAULT_PREVIEW_TIMEOUT_MS = 2_000 + +TabularSourceMode = Literal["live", "cached", "file_backed", "static"] +TabularHealthStatus = Literal["healthy", "warning", "error", "unknown"] +TabularDiagnosticSeverity = Literal["info", "warning", "error"] class TabularSourceError(ValueError): @@ -29,6 +35,10 @@ class TabularSourceValidationError(TabularSourceError): pass +class TabularSourceUnavailableError(TabularSourceError): + pass + + def parse_tabular_csv( csv_text: str, *, @@ -131,6 +141,32 @@ class TabularColumn: nullable: bool = True +@dataclass(frozen=True, slots=True) +class TabularPushdown: + projections: bool = False + pagination: bool = False + filters: tuple[str, ...] = () + aggregations: tuple[str, ...] = () + sorting: tuple[str, ...] = () + + +@dataclass(frozen=True, slots=True) +class TabularSourceHealth: + status: TabularHealthStatus = "unknown" + code: str = "source.health_unknown" + summary: str = "Source health has not been checked." + checked_at: datetime | None = None + details: Mapping[str, object] = field(default_factory=dict) + + +@dataclass(frozen=True, slots=True) +class TabularPreviewDiagnostic: + severity: TabularDiagnosticSeverity + code: str + message: str + details: Mapping[str, object] = field(default_factory=dict) + + @dataclass(frozen=True, slots=True) class TabularSource: """Opaque, policy-filtered source reference exposed to consuming modules.""" @@ -148,6 +184,9 @@ class TabularSource: updated_at: datetime | None = None capabilities: tuple[str, ...] = ("read",) metadata: Mapping[str, object] = field(default_factory=dict) + source_mode: TabularSourceMode = "cached" + pushdown: TabularPushdown = field(default_factory=TabularPushdown) + health: TabularSourceHealth = field(default_factory=TabularSourceHealth) @dataclass(frozen=True, slots=True) @@ -157,6 +196,8 @@ class TabularReadRequest: offset: int = 0 columns: tuple[str, ...] = () expected_fingerprint: str | None = None + max_bytes: int = DEFAULT_PREVIEW_BYTES + timeout_ms: int = DEFAULT_PREVIEW_TIMEOUT_MS @dataclass(frozen=True, slots=True) @@ -165,6 +206,12 @@ class TabularReadResult: rows: tuple[Mapping[str, object], ...] total_rows: int truncated: bool + returned_bytes: int = 0 + elapsed_ms: int = 0 + effective_row_limit: int = 0 + effective_byte_limit: int = 0 + effective_timeout_ms: int = 0 + diagnostics: tuple[TabularPreviewDiagnostic, ...] = () @dataclass(frozen=True, slots=True) @@ -247,7 +294,11 @@ def _capability(registry: object | None, name: str) -> object | None: __all__ = [ "CAPABILITY_CONNECTORS_TABULAR_SNAPSHOT_WRITER", "CAPABILITY_CONNECTORS_TABULAR_SOURCES", + "DEFAULT_PREVIEW_BYTES", + "DEFAULT_PREVIEW_TIMEOUT_MS", "TabularColumn", + "TabularPreviewDiagnostic", + "TabularPushdown", "TabularReadRequest", "TabularReadResult", "TabularSnapshotInput", @@ -257,6 +308,9 @@ __all__ = [ "TabularSourceError", "TabularSourceNotFoundError", "TabularSourceProvider", + "TabularSourceHealth", + "TabularSourceMode", + "TabularSourceUnavailableError", "TabularSourceValidationError", "parse_tabular_csv", "tabular_snapshot_writer", diff --git a/tests/test_tabular_source_contract.py b/tests/test_tabular_source_contract.py index 2ace5ba..7b58121 100644 --- a/tests/test_tabular_source_contract.py +++ b/tests/test_tabular_source_contract.py @@ -2,17 +2,25 @@ from __future__ import annotations import unittest +from govoplan_core.core.datasources import ( + DatasourceDescriptor, + DatasourceReadRequest, + DatasourceReadResult, +) from govoplan_core.core.modules import ModuleContext, ModuleManifest from govoplan_core.core.registry import PlatformRegistry from govoplan_core.core.tabular_sources import ( CAPABILITY_CONNECTORS_TABULAR_SNAPSHOT_WRITER, CAPABILITY_CONNECTORS_TABULAR_SOURCES, TabularColumn, + TabularPreviewDiagnostic, + TabularPushdown, TabularReadRequest, TabularReadResult, TabularSnapshotInput, TabularSnapshotWriter, TabularSource, + TabularSourceHealth, TabularSourceProvider, TabularSourceValidationError, parse_tabular_csv, @@ -30,6 +38,13 @@ class _TabularProvider: schema=(TabularColumn(name="case_id", data_type="string", nullable=False),), fingerprint="abc123", row_count=1, + source_mode="cached", + pushdown=TabularPushdown(projections=True, pagination=True), + health=TabularSourceHealth( + status="healthy", + code="snapshot.ready", + summary="Immutable snapshot is ready.", + ), ) def list_sources(self, session, principal, *, query="", limit=100): @@ -51,6 +66,18 @@ class _TabularProvider: rows=selected, total_rows=len(rows), truncated=len(selected) < len(rows), + returned_bytes=18, + elapsed_ms=1, + effective_row_limit=request.limit, + effective_byte_limit=request.max_bytes, + effective_timeout_ms=request.timeout_ms, + diagnostics=( + TabularPreviewDiagnostic( + severity="info", + code="preview.bounded", + message="The preview used explicit budgets.", + ), + ), ) def create_snapshot(self, session, principal, *, snapshot): @@ -96,8 +123,47 @@ class TabularSourceContractTests(unittest.TestCase): ) self.assertEqual(({"case_id": "A-1"},), result.rows) + self.assertEqual("cached", result.source.source_mode) + self.assertTrue(result.source.pushdown.projections) + self.assertEqual("healthy", result.source.health.status) + self.assertEqual("preview.bounded", result.diagnostics[0].code) + self.assertEqual(1_000_000, request.max_bytes) + self.assertEqual(2_000, request.timeout_ms) self.assertEqual(provider.source, provider.create_snapshot(object(), object(), snapshot=snapshot)) + def test_datasource_read_contract_preserves_live_preview_evidence(self) -> None: + request = DatasourceReadRequest(datasource_ref="datasource:monthly-cases") + result = DatasourceReadResult( + datasource=DatasourceDescriptor( + ref=request.datasource_ref, + source_name="monthly_cases", + name="Monthly cases", + kind="database", + mode="live", + shape="tabular", + ), + rows=(), + total_rows=0, + truncated=False, + returned_bytes=2, + elapsed_ms=3, + effective_row_limit=request.limit, + effective_byte_limit=request.max_bytes, + effective_timeout_ms=request.timeout_ms, + diagnostics=( + TabularPreviewDiagnostic( + severity="info", + code="preview.complete", + message="The bounded preview completed.", + ), + ), + ) + + self.assertEqual(1_000_000, request.max_bytes) + self.assertEqual(2_000, request.timeout_ms) + self.assertEqual(2, result.returned_bytes) + self.assertEqual("preview.complete", result.diagnostics[0].code) + def test_shared_csv_parser_preserves_identifier_zeroes_and_rejects_extra_values(self) -> None: rows = parse_tabular_csv( "case_id;amount;active\n0012;7.5;true\n\n",