Add provider-neutral tabular source contracts

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",
+]

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

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