From d36bb94335314e0527e21728872797222603675e Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Tue, 28 Jul 2026 11:12:24 +0200 Subject: [PATCH] Add provider-neutral tabular source contracts --- src/govoplan_core/core/tabular_sources.py | 230 ++++++++++++++++++++++ src/govoplan_core/settings.py | 2 +- tests/test_tabular_source_contract.py | 117 +++++++++++ 3 files changed, 348 insertions(+), 1 deletion(-) create mode 100644 src/govoplan_core/core/tabular_sources.py create mode 100644 tests/test_tabular_source_contract.py diff --git a/src/govoplan_core/core/tabular_sources.py b/src/govoplan_core/core/tabular_sources.py new file mode 100644 index 0000000..7d83dbc --- /dev/null +++ b/src/govoplan_core/core/tabular_sources.py @@ -0,0 +1,230 @@ +from __future__ import annotations + +import csv +import io +import re +from collections.abc import Mapping, Sequence +from dataclasses import dataclass, field +from datetime import datetime +from typing import Protocol, runtime_checkable + + +CAPABILITY_CONNECTORS_TABULAR_SOURCES = "connectors.tabularSources" +CAPABILITY_CONNECTORS_TABULAR_SNAPSHOT_WRITER = "connectors.tabularSnapshotWriter" + + +class TabularSourceError(ValueError): + """Stable base error for provider-neutral tabular source operations.""" + + +class TabularSourceNotFoundError(TabularSourceError): + pass + + +class TabularSourceAccessError(TabularSourceError): + pass + + +class TabularSourceValidationError(TabularSourceError): + pass + + +def parse_tabular_csv( + csv_text: str, + *, + delimiter: str = ",", + max_rows: int = 10_000, +) -> tuple[Mapping[str, object], ...]: + """Parse a bounded CSV document into JSON-compatible tabular rows.""" + + if len(delimiter) != 1: + raise TabularSourceValidationError("CSV delimiter must be one character.") + try: + reader = csv.DictReader(io.StringIO(csv_text), delimiter=delimiter) + if not reader.fieldnames or any(not str(name or "").strip() for name in reader.fieldnames): + raise TabularSourceValidationError("CSV input requires a non-empty header row.") + normalized_headers = [str(name).strip() for name in reader.fieldnames] + normalized_headers[0] = normalized_headers[0].lstrip("\ufeff") + if any(not name for name in normalized_headers): + raise TabularSourceValidationError("CSV input requires a non-empty header row.") + if len(set(normalized_headers)) != len(normalized_headers): + raise TabularSourceValidationError("CSV column names must be unique.") + rows: list[dict[str, object]] = [] + for row in reader: + extras = row.get(None) + if extras and any(str(value or "").strip() for value in extras): + raise TabularSourceValidationError( + "A CSV row contains more values than the header defines." + ) + values = [row.get(original) for original in reader.fieldnames] + if all(value is None or not value.strip() for value in values): + continue + if len(rows) >= max_rows: + raise TabularSourceValidationError( + f"CSV snapshots are limited to {max_rows:,} rows." + ) + rows.append( + { + normalized_headers[position]: _csv_scalar(row.get(original)) + for position, original in enumerate(reader.fieldnames) + } + ) + return tuple(rows) + except csv.Error as exc: + raise TabularSourceValidationError(f"CSV input could not be parsed: {exc}") from exc + + +def _csv_scalar(value: str | None) -> object: + if value is None: + return None + text = value.strip() + if not text: + return None + lowered = text.casefold() + if lowered in {"true", "false"}: + return lowered == "true" + if re.fullmatch(r"-?(?:0|[1-9][0-9]*)", text): + return int(text) + if re.fullmatch(r"-?(?:0|[1-9][0-9]*)\.[0-9]+", text): + return float(text) + return text + + +@dataclass(frozen=True, slots=True) +class TabularColumn: + name: str + data_type: str + nullable: bool = True + + +@dataclass(frozen=True, slots=True) +class TabularSource: + """Opaque, policy-filtered source reference exposed to consuming modules.""" + + ref: str + provider: str + source_name: str + name: str + description: str | None = None + schema: tuple[TabularColumn, ...] = () + schema_version: str = "1" + fingerprint: str = "" + row_count: int | None = None + byte_count: int | None = None + updated_at: datetime | None = None + capabilities: tuple[str, ...] = ("read",) + metadata: Mapping[str, object] = field(default_factory=dict) + + +@dataclass(frozen=True, slots=True) +class TabularReadRequest: + source_ref: str + limit: int = 250 + offset: int = 0 + columns: tuple[str, ...] = () + expected_fingerprint: str | None = None + + +@dataclass(frozen=True, slots=True) +class TabularReadResult: + source: TabularSource + rows: tuple[Mapping[str, object], ...] + total_rows: int + truncated: bool + + +@dataclass(frozen=True, slots=True) +class TabularSnapshotInput: + name: str + source_name: str + rows: tuple[Mapping[str, object], ...] + description: str | None = None + metadata: Mapping[str, object] = field(default_factory=dict) + + +@runtime_checkable +class TabularSourceProvider(Protocol): + """Principal-aware catalogue and bounded reader for tabular sources.""" + + def list_sources( + self, + session: object, + principal: object, + *, + query: str = "", + limit: int = 100, + ) -> Sequence[TabularSource]: + ... + + def get_source( + self, + session: object, + principal: object, + *, + source_ref: str, + ) -> TabularSource | None: + ... + + def read_source( + self, + session: object, + principal: object, + *, + request: TabularReadRequest, + ) -> TabularReadResult: + ... + + +@runtime_checkable +class TabularSnapshotWriter(Protocol): + """Optional capability for importing bounded static tabular snapshots.""" + + def create_snapshot( + self, + session: object, + principal: object, + *, + snapshot: TabularSnapshotInput, + ) -> TabularSource: + ... + + +def tabular_source_provider(registry: object | None) -> TabularSourceProvider | None: + capability = _capability(registry, CAPABILITY_CONNECTORS_TABULAR_SOURCES) + return capability if isinstance(capability, TabularSourceProvider) else None + + +def tabular_snapshot_writer(registry: object | None) -> TabularSnapshotWriter | None: + capability = _capability(registry, CAPABILITY_CONNECTORS_TABULAR_SNAPSHOT_WRITER) + return capability if isinstance(capability, TabularSnapshotWriter) else None + + +def _capability(registry: object | None, name: str) -> object | None: + if ( + registry is None + or not hasattr(registry, "has_capability") + or not hasattr(registry, "capability") + or not registry.has_capability(name) + ): + return None + return registry.capability(name) + + +__all__ = [ + "CAPABILITY_CONNECTORS_TABULAR_SNAPSHOT_WRITER", + "CAPABILITY_CONNECTORS_TABULAR_SOURCES", + "TabularColumn", + "TabularReadRequest", + "TabularReadResult", + "TabularSnapshotInput", + "TabularSnapshotWriter", + "TabularSource", + "TabularSourceAccessError", + "TabularSourceError", + "TabularSourceNotFoundError", + "TabularSourceProvider", + "TabularSourceValidationError", + "parse_tabular_csv", + "tabular_snapshot_writer", + "tabular_source_provider", +] diff --git a/src/govoplan_core/settings.py b/src/govoplan_core/settings.py index 96faf35..da65178 100644 --- a/src/govoplan_core/settings.py +++ b/src/govoplan_core/settings.py @@ -17,7 +17,7 @@ class Settings(BaseSettings): tenant_data_mode: str = Field(default="shared", alias="TENANT_DATA_MODE") tenant_db_url_template: str | None = Field(default=None, alias="TENANT_DB_URL_TEMPLATE") tenant_schema_template: str | None = Field(default=None, alias="TENANT_SCHEMA_TEMPLATE") - enabled_modules: str = Field(default="tenancy,organizations,identity,access,admin,dashboard,policy,audit,campaigns,files,mail,calendar,poll,scheduling,dataflow,notifications,docs,ops", alias="ENABLED_MODULES") + enabled_modules: str = Field(default="tenancy,organizations,identity,access,admin,dashboard,policy,audit,campaigns,files,mail,calendar,poll,scheduling,connectors,dataflow,notifications,docs,ops", alias="ENABLED_MODULES") migration_track: str = Field(default="release", alias="GOVOPLAN_MIGRATION_TRACK") redis_url: str = Field(default="redis://redis:6379/0", alias="REDIS_URL") celery_enabled: bool = Field(default=False, alias="CELERY_ENABLED") diff --git a/tests/test_tabular_source_contract.py b/tests/test_tabular_source_contract.py new file mode 100644 index 0000000..2ace5ba --- /dev/null +++ b/tests/test_tabular_source_contract.py @@ -0,0 +1,117 @@ +from __future__ import annotations + +import unittest + +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, + TabularReadRequest, + TabularReadResult, + TabularSnapshotInput, + TabularSnapshotWriter, + TabularSource, + TabularSourceProvider, + TabularSourceValidationError, + parse_tabular_csv, + tabular_snapshot_writer, + tabular_source_provider, +) + + +class _TabularProvider: + source = TabularSource( + ref="snapshot:source-1", + provider="snapshot", + source_name="monthly_cases", + name="Monthly cases", + schema=(TabularColumn(name="case_id", data_type="string", nullable=False),), + fingerprint="abc123", + row_count=1, + ) + + def list_sources(self, session, principal, *, query="", limit=100): + del session, principal + if query and query.casefold() not in self.source.name.casefold(): + return () + return (self.source,)[:limit] + + def get_source(self, session, principal, *, source_ref): + del session, principal + return self.source if source_ref == self.source.ref else None + + def read_source(self, session, principal, *, request): + del session, principal + rows = ({"case_id": "A-1"},) + selected = rows[request.offset : request.offset + request.limit] + return TabularReadResult( + source=self.source, + rows=selected, + total_rows=len(rows), + truncated=len(selected) < len(rows), + ) + + def create_snapshot(self, session, principal, *, snapshot): + del session, principal, snapshot + return self.source + + +class TabularSourceContractTests(unittest.TestCase): + def test_provider_and_snapshot_writer_are_runtime_checkable(self) -> None: + provider = _TabularProvider() + + self.assertIsInstance(provider, TabularSourceProvider) + self.assertIsInstance(provider, TabularSnapshotWriter) + + def test_capabilities_resolve_without_importing_connectors(self) -> None: + provider = _TabularProvider() + registry = PlatformRegistry() + registry.register( + ModuleManifest( + id="tabular_contract_test", + name="Tabular contract test", + version="test", + capability_factories={ + CAPABILITY_CONNECTORS_TABULAR_SOURCES: lambda context: provider, + CAPABILITY_CONNECTORS_TABULAR_SNAPSHOT_WRITER: lambda context: provider, + }, + ) + ) + registry.configure_capability_context(ModuleContext(registry=registry, settings=object())) + + self.assertIs(provider, tabular_source_provider(registry)) + self.assertIs(provider, tabular_snapshot_writer(registry)) + self.assertIsNone(tabular_source_provider(PlatformRegistry())) + + def test_read_and_snapshot_dtos_are_immutable_and_bounded_by_callers(self) -> None: + provider = _TabularProvider() + request = TabularReadRequest(source_ref=provider.source.ref, limit=1) + result = provider.read_source(object(), object(), request=request) + snapshot = TabularSnapshotInput( + name="Monthly cases", + source_name="monthly_cases", + rows=({"case_id": "A-1"},), + ) + + self.assertEqual(({"case_id": "A-1"},), result.rows) + self.assertEqual(provider.source, provider.create_snapshot(object(), object(), snapshot=snapshot)) + + 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", + delimiter=";", + max_rows=2, + ) + + self.assertEqual( + ({"case_id": "0012", "amount": 7.5, "active": True},), + rows, + ) + with self.assertRaises(TabularSourceValidationError): + parse_tabular_csv("id,name\n1,Ada,extra\n") + + +if __name__ == "__main__": + unittest.main()