Implement governed tabular source snapshots

README.md

@@ -4,14 +4,32 @@
 **Repository type:** connector (connector-hub).
 <!-- govoplan-repository-type:end -->
 
-`govoplan-connectors` will own integration catalogues and generic external
+`govoplan-connectors` owns integration catalogues and generic external system
-system connection patterns for GovOPlaN.
+connection patterns for GovOPlaN.
 
 The module should make external systems discoverable, testable, and usable
 without taking ownership of their business semantics. Domain-specific modules
 remain responsible for case, file, workflow, payment, mail, identity, document,
 or reporting behavior.
 
+## Executable First Slice
+
+The first executable connector capability provides tenant-isolated tabular
+sources for Dataflow. Operators can import bounded JSON or CSV snapshots,
+inspect inferred schemas, and expose immutable source references and content
+fingerprints through `connectors.tabular_sources@0.1.0`.
+
+Database, API, file, and warehouse providers can implement the same core
+capability later. Dataflow stores opaque references and expected fingerprints;
+it never stores connector credentials.
+
+Development:
+
+```bash
+/mnt/DATA/git/govoplan/.venv/bin/python -m pip install -e .
+/mnt/DATA/git/govoplan/.venv/bin/python -m unittest discover -s tests
+```
+
 See:
 
 - [Connector concept](docs/CONCEPT.md)

pyproject.toml

@@ -0,0 +1,22 @@
+[build-system]
+requires = ["setuptools>=69", "wheel"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "govoplan-connectors"
+version = "0.1.14"
+description = "Governed connector catalogue and tabular source capabilities for GovOPlaN."
+readme = "README.md"
+requires-python = ">=3.12"
+license = "AGPL-3.0-or-later"
+authors = [{ name = "GovOPlaN" }]
+dependencies = ["govoplan-core>=0.1.14"]
+
+[tool.setuptools.packages.find]
+where = ["src"]
+
+[tool.setuptools.package-data]
+govoplan_connectors = ["py.typed"]
+
+[project.entry-points."govoplan.modules"]
+connectors = "govoplan_connectors.backend.manifest:get_manifest"

src/govoplan_connectors/__init__.py

@@ -0,0 +1,3 @@
+"""GovOPlaN Connectors module."""
+
+__all__: list[str] = []

src/govoplan_connectors/backend/__init__.py

@@ -0,0 +1 @@
+"""Connector backend package."""

src/govoplan_connectors/backend/db/__init__.py

@@ -0,0 +1,3 @@
+from govoplan_connectors.backend.db.models import ConnectorTabularSource
+
+__all__ = ["ConnectorTabularSource"]

src/govoplan_connectors/backend/db/models.py

@@ -0,0 +1,44 @@
+from __future__ import annotations
+
+import uuid
+from datetime import datetime
+from typing import Any
+
+from sqlalchemy import DateTime, Index, Integer, JSON, String, Text, UniqueConstraint
+from sqlalchemy.orm import Mapped, mapped_column
+
+from govoplan_core.db.base import Base, TimestampMixin
+
+
+def new_uuid() -> str:
+    return str(uuid.uuid4())
+
+
+class ConnectorTabularSource(Base, TimestampMixin):
+    __tablename__ = "connector_tabular_sources"
+    __table_args__ = (
+        UniqueConstraint("tenant_id", "source_name", name="uq_connector_tabular_source_name"),
+        Index("ix_connector_tabular_sources_tenant_status", "tenant_id", "status"),
+        Index("ix_connector_tabular_sources_tenant_updated", "tenant_id", "updated_at"),
+    )
+
+    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid)
+    tenant_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True)
+    provider: Mapped[str] = mapped_column(String(50), default="snapshot", nullable=False, index=True)
+    source_name: Mapped[str] = mapped_column(String(120), nullable=False)
+    name: Mapped[str] = mapped_column(String(300), nullable=False)
+    description: Mapped[str | None] = mapped_column(Text)
+    status: Mapped[str] = mapped_column(String(30), default="active", nullable=False, index=True)
+    schema_version: Mapped[int] = mapped_column(Integer, default=1, nullable=False)
+    schema_: Mapped[list[dict[str, Any]]] = mapped_column("schema", JSON, default=list, nullable=False)
+    rows: Mapped[list[dict[str, Any]]] = mapped_column(JSON, default=list, nullable=False)
+    fingerprint: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
+    row_count: Mapped[int] = mapped_column(Integer, nullable=False)
+    byte_count: Mapped[int] = mapped_column(Integer, nullable=False)
+    metadata_: Mapped[dict[str, Any]] = mapped_column("metadata", JSON, default=dict, nullable=False)
+    created_by: Mapped[str | None] = mapped_column(String(255), nullable=True, index=True)
+    updated_by: Mapped[str | None] = mapped_column(String(255), nullable=True, index=True)
+    deleted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True, index=True)
+
+
+__all__ = ["ConnectorTabularSource", "new_uuid"]

src/govoplan_connectors/backend/manifest.py

@@ -0,0 +1,190 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+from govoplan_core.core.access import (
+    CAPABILITY_AUTH_PERMISSION_EVALUATOR,
+    CAPABILITY_AUTH_PRINCIPAL_RESOLVER,
+)
+from govoplan_core.core.module_guards import (
+    drop_table_retirement_provider,
+    persistent_table_uninstall_guard,
+)
+from govoplan_core.core.modules import (
+    DocumentationTopic,
+    MigrationSpec,
+    ModuleInterfaceProvider,
+    ModuleManifest,
+    PermissionDefinition,
+    RoleTemplate,
+)
+from govoplan_core.core.tabular_sources import (
+    CAPABILITY_CONNECTORS_TABULAR_SNAPSHOT_WRITER,
+    CAPABILITY_CONNECTORS_TABULAR_SOURCES,
+)
+from govoplan_core.db.base import Base
+from govoplan_connectors.backend.db.models import ConnectorTabularSource
+from govoplan_connectors.backend.tabular_sources import (
+    ADMIN_SCOPE,
+    READ_SCOPE,
+    WRITE_SCOPE,
+    SqlTabularSourceProvider,
+)
+
+
+MODULE_ID = "connectors"
+MODULE_VERSION = "0.1.14"
+TABULAR_SOURCE_INTERFACE_VERSION = "0.1.0"
+
+
+def _permission(scope: str, label: str, description: str) -> PermissionDefinition:
+    module_id, resource, action = scope.split(":", 2)
+    return PermissionDefinition(
+        scope=scope,
+        label=label,
+        description=description,
+        category="Connectors",
+        level="tenant",
+        module_id=module_id,
+        resource=resource,
+        action=action,
+    )
+
+
+PERMISSIONS = (
+    _permission(
+        READ_SCOPE,
+        "View tabular sources",
+        "Discover and preview policy-visible tabular connector sources.",
+    ),
+    _permission(
+        WRITE_SCOPE,
+        "Manage tabular sources",
+        "Import and retire bounded tabular snapshots.",
+    ),
+    _permission(
+        ADMIN_SCOPE,
+        "Administer connector sources",
+        "Manage every tenant connector source and future source policies.",
+    ),
+)
+
+ROLE_TEMPLATES = (
+    RoleTemplate(
+        slug="connector_source_manager",
+        name="Connector source manager",
+        description="Discover, import, preview, and retire tabular sources.",
+        permissions=(READ_SCOPE, WRITE_SCOPE),
+    ),
+    RoleTemplate(
+        slug="connector_source_reader",
+        name="Connector source reader",
+        description="Discover and preview tabular connector sources.",
+        permissions=(READ_SCOPE,),
+    ),
+)
+
+
+def _router(_context):
+    from govoplan_connectors.backend.router import router
+
+    return router
+
+
+def _provider(_context) -> SqlTabularSourceProvider:
+    return SqlTabularSourceProvider()
+
+
+def _tenant_summary(session, tenant_id: str) -> dict[str, int]:
+    return {
+        "connector_tabular_sources": (
+            session.query(ConnectorTabularSource)
+            .filter(
+                ConnectorTabularSource.tenant_id == tenant_id,
+                ConnectorTabularSource.deleted_at.is_(None),
+            )
+            .count()
+        )
+    }
+
+
+manifest = ModuleManifest(
+    id=MODULE_ID,
+    name="Connectors",
+    version=MODULE_VERSION,
+    optional_dependencies=("access", "audit", "files", "policy"),
+    required_capabilities=(
+        CAPABILITY_AUTH_PRINCIPAL_RESOLVER,
+        CAPABILITY_AUTH_PERMISSION_EVALUATOR,
+    ),
+    provides_interfaces=(
+        ModuleInterfaceProvider(
+            name="connectors.tabular_sources",
+            version=TABULAR_SOURCE_INTERFACE_VERSION,
+        ),
+        ModuleInterfaceProvider(
+            name="connectors.tabular_snapshot_writer",
+            version=TABULAR_SOURCE_INTERFACE_VERSION,
+        ),
+    ),
+    permissions=PERMISSIONS,
+    role_templates=ROLE_TEMPLATES,
+    route_factory=_router,
+    capability_factories={
+        CAPABILITY_CONNECTORS_TABULAR_SOURCES: _provider,
+        CAPABILITY_CONNECTORS_TABULAR_SNAPSHOT_WRITER: _provider,
+    },
+    tenant_summary_providers=(_tenant_summary,),
+    migration_spec=MigrationSpec(
+        module_id=MODULE_ID,
+        metadata=Base.metadata,
+        script_location=str(Path(__file__).with_name("migrations") / "versions"),
+        retirement_supported=True,
+        retirement_provider=drop_table_retirement_provider(
+            ConnectorTabularSource,
+            label="Connectors",
+        ),
+        retirement_notes=(
+            "Destructive retirement drops connector-owned source snapshots after "
+            "the installer captures a database snapshot."
+        ),
+    ),
+    uninstall_guard_providers=(
+        persistent_table_uninstall_guard(
+            ConnectorTabularSource,
+            label="Connectors",
+        ),
+    ),
+    documentation=(
+        DocumentationTopic(
+            id="connectors.tabular-sources",
+            title="Governed tabular sources",
+            summary="Provider-neutral source discovery and bounded reads for Dataflow.",
+            body=(
+                "Connectors owns source configuration, access checks, schema discovery, "
+                "fingerprints, and bounded reads. Dataflow stores only opaque source "
+                "references and expected fingerprints. The first executable provider "
+                "imports immutable JSON or CSV snapshots; database and API providers "
+                "can implement the same capability without changing Dataflow."
+            ),
+            layer="available",
+            documentation_types=("admin", "user"),
+            audience=("operator", "module_admin", "power_user"),
+            related_modules=("dataflow", "files", "reporting", "risk_compliance"),
+            order=40,
+        ),
+    ),
+)
+
+
+def get_manifest() -> ModuleManifest:
+    return manifest
+
+
+__all__ = [
+    "MODULE_ID",
+    "MODULE_VERSION",
+    "TABULAR_SOURCE_INTERFACE_VERSION",
+    "get_manifest",
+    "manifest",
+]

src/govoplan_connectors/backend/migrations/__init__.py

@@ -0,0 +1 @@
+"""Connectors migrations."""

src/govoplan_connectors/backend/migrations/versions/__init__.py

@@ -0,0 +1 @@
+"""Connectors migration revisions."""

src/govoplan_connectors/backend/migrations/versions/e6b7c8d9f0a1_v0114_connectors_baseline.py

@@ -0,0 +1,141 @@
+"""v0.1.14 Connectors baseline
+
+Revision ID: e6b7c8d9f0a1
+Revises: None
+Create Date: 2026-07-28 00:00:00.000000
+"""
+from __future__ import annotations
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision = "e6b7c8d9f0a1"
+down_revision = None
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "connector_tabular_sources",
+        sa.Column("id", sa.String(length=36), nullable=False),
+        sa.Column("tenant_id", sa.String(length=36), nullable=False),
+        sa.Column("provider", sa.String(length=50), nullable=False),
+        sa.Column("source_name", sa.String(length=120), nullable=False),
+        sa.Column("name", sa.String(length=300), nullable=False),
+        sa.Column("description", sa.Text(), nullable=True),
+        sa.Column("status", sa.String(length=30), nullable=False),
+        sa.Column("schema_version", sa.Integer(), nullable=False),
+        sa.Column("schema", sa.JSON(), nullable=False),
+        sa.Column("rows", sa.JSON(), nullable=False),
+        sa.Column("fingerprint", sa.String(length=64), nullable=False),
+        sa.Column("row_count", sa.Integer(), nullable=False),
+        sa.Column("byte_count", sa.Integer(), nullable=False),
+        sa.Column("metadata", sa.JSON(), nullable=False),
+        sa.Column("created_by", sa.String(length=255), nullable=True),
+        sa.Column("updated_by", sa.String(length=255), nullable=True),
+        sa.Column("deleted_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+        sa.PrimaryKeyConstraint("id", name=op.f("pk_connector_tabular_sources")),
+        sa.UniqueConstraint(
+            "tenant_id",
+            "source_name",
+            name="uq_connector_tabular_source_name",
+        ),
+    )
+    op.create_index(
+        op.f("ix_connector_tabular_sources_created_by"),
+        "connector_tabular_sources",
+        ["created_by"],
+        unique=False,
+    )
+    op.create_index(
+        op.f("ix_connector_tabular_sources_deleted_at"),
+        "connector_tabular_sources",
+        ["deleted_at"],
+        unique=False,
+    )
+    op.create_index(
+        op.f("ix_connector_tabular_sources_fingerprint"),
+        "connector_tabular_sources",
+        ["fingerprint"],
+        unique=False,
+    )
+    op.create_index(
+        op.f("ix_connector_tabular_sources_provider"),
+        "connector_tabular_sources",
+        ["provider"],
+        unique=False,
+    )
+    op.create_index(
+        op.f("ix_connector_tabular_sources_status"),
+        "connector_tabular_sources",
+        ["status"],
+        unique=False,
+    )
+    op.create_index(
+        op.f("ix_connector_tabular_sources_tenant_id"),
+        "connector_tabular_sources",
+        ["tenant_id"],
+        unique=False,
+    )
+    op.create_index(
+        op.f("ix_connector_tabular_sources_updated_by"),
+        "connector_tabular_sources",
+        ["updated_by"],
+        unique=False,
+    )
+    op.create_index(
+        "ix_connector_tabular_sources_tenant_status",
+        "connector_tabular_sources",
+        ["tenant_id", "status"],
+        unique=False,
+    )
+    op.create_index(
+        "ix_connector_tabular_sources_tenant_updated",
+        "connector_tabular_sources",
+        ["tenant_id", "updated_at"],
+        unique=False,
+    )
+
+
+def downgrade() -> None:
+    op.drop_index(
+        "ix_connector_tabular_sources_tenant_updated",
+        table_name="connector_tabular_sources",
+    )
+    op.drop_index(
+        "ix_connector_tabular_sources_tenant_status",
+        table_name="connector_tabular_sources",
+    )
+    op.drop_index(
+        op.f("ix_connector_tabular_sources_updated_by"),
+        table_name="connector_tabular_sources",
+    )
+    op.drop_index(
+        op.f("ix_connector_tabular_sources_tenant_id"),
+        table_name="connector_tabular_sources",
+    )
+    op.drop_index(
+        op.f("ix_connector_tabular_sources_status"),
+        table_name="connector_tabular_sources",
+    )
+    op.drop_index(
+        op.f("ix_connector_tabular_sources_provider"),
+        table_name="connector_tabular_sources",
+    )
+    op.drop_index(
+        op.f("ix_connector_tabular_sources_fingerprint"),
+        table_name="connector_tabular_sources",
+    )
+    op.drop_index(
+        op.f("ix_connector_tabular_sources_deleted_at"),
+        table_name="connector_tabular_sources",
+    )
+    op.drop_index(
+        op.f("ix_connector_tabular_sources_created_by"),
+        table_name="connector_tabular_sources",
+    )
+    op.drop_table("connector_tabular_sources")

src/govoplan_connectors/backend/router.py

@@ -0,0 +1,211 @@
+from __future__ import annotations
+
+from fastapi import APIRouter, Depends, HTTPException, Query, status
+from sqlalchemy.orm import Session
+
+from govoplan_core.audit.logging import audit_event
+from govoplan_core.auth import ApiPrincipal, get_api_principal, has_scope
+from govoplan_core.core.tabular_sources import (
+    TabularReadRequest,
+    TabularSnapshotInput,
+    TabularSource,
+    TabularSourceAccessError,
+    TabularSourceError,
+    TabularSourceNotFoundError,
+)
+from govoplan_core.db.session import get_session
+from govoplan_connectors.backend.schemas import (
+    SnapshotCreateRequest,
+    TabularColumnResponse,
+    TabularSourceDeleteResponse,
+    TabularSourceListResponse,
+    TabularSourcePreviewResponse,
+    TabularSourceResponse,
+)
+from govoplan_connectors.backend.tabular_sources import (
+    ADMIN_SCOPE,
+    READ_SCOPE,
+    WRITE_SCOPE,
+    SqlTabularSourceProvider,
+    parse_csv_snapshot,
+)
+
+
+router = APIRouter(prefix="/connectors", tags=["connectors"])
+provider = SqlTabularSourceProvider()
+
+
+def _require_any_scope(principal: ApiPrincipal, *scopes: str) -> None:
+    if any(has_scope(principal, scope) for scope in scopes):
+        return
+    raise HTTPException(
+        status_code=status.HTTP_403_FORBIDDEN,
+        detail=f"Missing one of the required scopes: {', '.join(scopes)}",
+    )
+
+
+def _http_error(exc: TabularSourceError) -> HTTPException:
+    if isinstance(exc, TabularSourceNotFoundError):
+        return HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc))
+    if isinstance(exc, TabularSourceAccessError):
+        return HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=str(exc))
+    return HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc))
+
+
+@router.get("/tabular-sources", response_model=TabularSourceListResponse)
+def api_list_tabular_sources(
+    query: str = Query(default="", max_length=200),
+    limit: int = Query(default=100, ge=1, le=100),
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(get_api_principal),
+) -> TabularSourceListResponse:
+    _require_any_scope(principal, READ_SCOPE, ADMIN_SCOPE)
+    sources = provider.list_sources(
+        session,
+        principal,
+        query=query,
+        limit=limit,
+    )
+    return TabularSourceListResponse(sources=[_source_response(source) for source in sources])
+
+
+@router.post(
+    "/tabular-sources/snapshots",
+    response_model=TabularSourceResponse,
+    status_code=status.HTTP_201_CREATED,
+)
+def api_create_tabular_snapshot(
+    payload: SnapshotCreateRequest,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(get_api_principal),
+) -> TabularSourceResponse:
+    _require_any_scope(principal, WRITE_SCOPE, ADMIN_SCOPE)
+    try:
+        rows = (
+            tuple(payload.rows or ())
+            if payload.format == "json"
+            else parse_csv_snapshot(payload.csv_text or "", delimiter=payload.delimiter)
+        )
+        source = provider.create_snapshot(
+            session,
+            principal,
+            snapshot=TabularSnapshotInput(
+                name=payload.name,
+                source_name=payload.source_name,
+                description=payload.description,
+                rows=rows,
+                metadata={"import_format": payload.format},
+            ),
+        )
+    except TabularSourceError as exc:
+        raise _http_error(exc) from exc
+    audit_event(
+        session,
+        tenant_id=principal.tenant_id,
+        user_id=getattr(principal.user, "id", None),
+        api_key_id=principal.api_key_id,
+        action="connectors.tabular_snapshot.created",
+        object_type="connector_tabular_source",
+        object_id=source.ref,
+        details={
+            "provider": source.provider,
+            "source_name": source.source_name,
+            "fingerprint": source.fingerprint,
+            "row_count": source.row_count,
+        },
+    )
+    session.commit()
+    return _source_response(source)
+
+
+@router.get(
+    "/tabular-sources/{source_id}/preview",
+    response_model=TabularSourcePreviewResponse,
+)
+def api_preview_tabular_source(
+    source_id: str,
+    limit: int = Query(default=100, ge=1, le=500),
+    offset: int = Query(default=0, ge=0),
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(get_api_principal),
+) -> TabularSourcePreviewResponse:
+    _require_any_scope(principal, READ_SCOPE, ADMIN_SCOPE)
+    try:
+        result = provider.read_source(
+            session,
+            principal,
+            request=TabularReadRequest(
+                source_ref=f"snapshot:{source_id}",
+                limit=limit,
+                offset=offset,
+            ),
+        )
+    except TabularSourceError as exc:
+        raise _http_error(exc) from exc
+    return TabularSourcePreviewResponse(
+        source=_source_response(result.source),
+        rows=[dict(row) for row in result.rows],
+        total_rows=result.total_rows,
+        truncated=result.truncated,
+    )
+
+
+@router.delete(
+    "/tabular-sources/{source_id}",
+    response_model=TabularSourceDeleteResponse,
+)
+def api_delete_tabular_source(
+    source_id: str,
+    session: Session = Depends(get_session),
+    principal: ApiPrincipal = Depends(get_api_principal),
+) -> TabularSourceDeleteResponse:
+    _require_any_scope(principal, WRITE_SCOPE, ADMIN_SCOPE)
+    source_ref = f"snapshot:{source_id}"
+    try:
+        source = provider.delete_snapshot(
+            session,
+            principal,
+            source_ref=source_ref,
+        )
+    except TabularSourceError as exc:
+        raise _http_error(exc) from exc
+    audit_event(
+        session,
+        tenant_id=principal.tenant_id,
+        user_id=getattr(principal.user, "id", None),
+        api_key_id=principal.api_key_id,
+        action="connectors.tabular_snapshot.deleted",
+        object_type="connector_tabular_source",
+        object_id=source_ref,
+        details={"source_name": source.source_name, "fingerprint": source.fingerprint},
+    )
+    session.commit()
+    return TabularSourceDeleteResponse(deleted=True, source_ref=source_ref)
+
+
+def _source_response(source: TabularSource) -> TabularSourceResponse:
+    return TabularSourceResponse(
+        ref=source.ref,
+        provider=source.provider,
+        source_name=source.source_name,
+        name=source.name,
+        description=source.description,
+        columns=[
+            TabularColumnResponse(
+                name=column.name,
+                data_type=column.data_type,
+                nullable=column.nullable,
+            )
+            for column in source.schema
+        ],
+        schema_version=source.schema_version,
+        fingerprint=source.fingerprint,
+        row_count=source.row_count,
+        byte_count=source.byte_count,
+        updated_at=source.updated_at.isoformat() if source.updated_at else None,
+        capabilities=list(source.capabilities),
+        metadata=dict(source.metadata),
+    )
+
+
+__all__ = ["router"]

src/govoplan_connectors/backend/schemas.py

@@ -0,0 +1,79 @@
+from __future__ import annotations
+
+from typing import Any, Literal
+
+from pydantic import BaseModel, Field, model_validator
+
+
+class SnapshotCreateRequest(BaseModel):
+    name: str = Field(min_length=1, max_length=300)
+    source_name: str = Field(
+        min_length=1,
+        max_length=120,
+        pattern=r"^[A-Za-z_][A-Za-z0-9_]*$",
+    )
+    description: str | None = Field(default=None, max_length=4000)
+    format: Literal["json", "csv"] = "json"
+    rows: list[dict[str, Any]] | None = Field(default=None, max_length=10_000)
+    csv_text: str | None = Field(default=None, max_length=5_000_000)
+    delimiter: Literal[",", ";", "\t", "|"] = ","
+
+    @model_validator(mode="after")
+    def validate_payload(self) -> "SnapshotCreateRequest":
+        if self.format == "json" and self.rows is None:
+            raise ValueError("JSON snapshots require rows.")
+        if self.format == "json" and self.csv_text is not None:
+            raise ValueError("JSON snapshots cannot include CSV text.")
+        if self.format == "csv" and not self.csv_text:
+            raise ValueError("CSV snapshots require CSV text.")
+        if self.format == "csv" and self.rows is not None:
+            raise ValueError("CSV snapshots cannot include JSON rows.")
+        return self
+
+
+class TabularColumnResponse(BaseModel):
+    name: str
+    data_type: str
+    nullable: bool
+
+
+class TabularSourceResponse(BaseModel):
+    ref: str
+    provider: str
+    source_name: str
+    name: str
+    description: str | None
+    columns: list[TabularColumnResponse]
+    schema_version: str
+    fingerprint: str
+    row_count: int | None
+    byte_count: int | None
+    updated_at: str | None
+    capabilities: list[str]
+    metadata: dict[str, Any]
+
+
+class TabularSourceListResponse(BaseModel):
+    sources: list[TabularSourceResponse]
+
+
+class TabularSourcePreviewResponse(BaseModel):
+    source: TabularSourceResponse
+    rows: list[dict[str, Any]]
+    total_rows: int
+    truncated: bool
+
+
+class TabularSourceDeleteResponse(BaseModel):
+    deleted: bool
+    source_ref: str
+
+
+__all__ = [
+    "SnapshotCreateRequest",
+    "TabularColumnResponse",
+    "TabularSourceDeleteResponse",
+    "TabularSourceListResponse",
+    "TabularSourcePreviewResponse",
+    "TabularSourceResponse",
+]

src/govoplan_connectors/backend/tabular_sources.py

@@ -0,0 +1,365 @@
+from __future__ import annotations
+
+import hashlib
+import json
+from collections.abc import Mapping, Sequence
+from datetime import datetime
+from decimal import Decimal
+from typing import Any
+
+from sqlalchemy import or_, select
+from sqlalchemy.orm import Session
+
+from govoplan_core.auth import ApiPrincipal, has_scope
+from govoplan_core.core.tabular_sources import (
+    TabularColumn,
+    TabularReadRequest,
+    TabularReadResult,
+    TabularSnapshotInput,
+    TabularSource,
+    TabularSourceAccessError,
+    TabularSourceNotFoundError,
+    TabularSourceValidationError,
+    parse_tabular_csv,
+)
+from govoplan_core.db.base import utcnow
+from govoplan_connectors.backend.db.models import ConnectorTabularSource
+
+
+READ_SCOPE = "connectors:source:read"
+WRITE_SCOPE = "connectors:source:write"
+ADMIN_SCOPE = "connectors:source:admin"
+MAX_SNAPSHOT_ROWS = 10_000
+MAX_SNAPSHOT_BYTES = 5_000_000
+MAX_READ_ROWS = 500
+
+
+class SqlTabularSourceProvider:
+    def list_sources(
+        self,
+        session: object,
+        principal: object,
+        *,
+        query: str = "",
+        limit: int = 100,
+    ) -> Sequence[TabularSource]:
+        db, api_principal = _context(session, principal, READ_SCOPE)
+        normalized_query = str(query or "").strip()
+        statement = select(ConnectorTabularSource).where(
+            ConnectorTabularSource.tenant_id == api_principal.tenant_id,
+            ConnectorTabularSource.deleted_at.is_(None),
+            ConnectorTabularSource.status == "active",
+        )
+        if normalized_query:
+            pattern = f"%{_escape_like(normalized_query)}%"
+            statement = statement.where(
+                or_(
+                    ConnectorTabularSource.name.ilike(pattern, escape="\\"),
+                    ConnectorTabularSource.source_name.ilike(pattern, escape="\\"),
+                )
+            )
+        statement = statement.order_by(
+            ConnectorTabularSource.updated_at.desc(),
+            ConnectorTabularSource.name,
+        ).limit(max(1, min(int(limit), 100)))
+        return tuple(_source_dto(item) for item in db.scalars(statement))
+
+    def get_source(
+        self,
+        session: object,
+        principal: object,
+        *,
+        source_ref: str,
+    ) -> TabularSource | None:
+        db, api_principal = _context(session, principal, READ_SCOPE)
+        item = _source_record(
+            db,
+            tenant_id=api_principal.tenant_id,
+            source_ref=source_ref,
+        )
+        return _source_dto(item) if item is not None else None
+
+    def read_source(
+        self,
+        session: object,
+        principal: object,
+        *,
+        request: TabularReadRequest,
+    ) -> TabularReadResult:
+        db, api_principal = _context(session, principal, READ_SCOPE)
+        item = _source_record(
+            db,
+            tenant_id=api_principal.tenant_id,
+            source_ref=request.source_ref,
+        )
+        if item is None:
+            raise TabularSourceNotFoundError("Tabular source not found.")
+        if request.expected_fingerprint and request.expected_fingerprint != item.fingerprint:
+            raise TabularSourceValidationError(
+                "The source fingerprint changed; refresh the source node before running it."
+            )
+
+        limit = max(1, min(int(request.limit), MAX_READ_ROWS))
+        offset = max(0, int(request.offset))
+        selected_columns = tuple(dict.fromkeys(request.columns))
+        known_columns = {column["name"] for column in item.schema_}
+        unknown_columns = [column for column in selected_columns if column not in known_columns]
+        if unknown_columns:
+            raise TabularSourceValidationError(
+                f"Unknown source columns: {', '.join(unknown_columns)}"
+            )
+        window = item.rows[offset : offset + limit]
+        rows = tuple(
+            {
+                key: value
+                for key, value in row.items()
+                if not selected_columns or key in selected_columns
+            }
+            for row in window
+        )
+        return TabularReadResult(
+            source=_source_dto(item),
+            rows=rows,
+            total_rows=item.row_count,
+            truncated=offset + len(rows) < item.row_count,
+        )
+
+    def create_snapshot(
+        self,
+        session: object,
+        principal: object,
+        *,
+        snapshot: TabularSnapshotInput,
+    ) -> TabularSource:
+        db, api_principal = _context(session, principal, WRITE_SCOPE)
+        name = snapshot.name.strip()
+        source_name = snapshot.source_name.strip()
+        if not name:
+            raise TabularSourceValidationError("Snapshot name is required.")
+        if not source_name:
+            raise TabularSourceValidationError("Snapshot source name is required.")
+        if len(snapshot.rows) > MAX_SNAPSHOT_ROWS:
+            raise TabularSourceValidationError(
+                f"Snapshots are limited to {MAX_SNAPSHOT_ROWS:,} rows."
+            )
+        rows = [_json_row(row) for row in snapshot.rows]
+        encoded = json.dumps(rows, sort_keys=True, separators=(",", ":"), default=str).encode("utf-8")
+        if len(encoded) > MAX_SNAPSHOT_BYTES:
+            raise TabularSourceValidationError(
+                f"Snapshots are limited to {MAX_SNAPSHOT_BYTES // 1_000_000} MB."
+            )
+        existing = db.scalar(
+            select(ConnectorTabularSource.id).where(
+                ConnectorTabularSource.tenant_id == api_principal.tenant_id,
+                ConnectorTabularSource.source_name == source_name,
+            )
+        )
+        if existing is not None:
+            raise TabularSourceValidationError(
+                f"A tabular source named {source_name!r} already exists."
+            )
+        schema = infer_schema(rows)
+        fingerprint = snapshot_fingerprint(rows, schema)
+        actor_id = _actor_id(api_principal)
+        item = ConnectorTabularSource(
+            tenant_id=api_principal.tenant_id,
+            provider="snapshot",
+            source_name=source_name,
+            name=name,
+            description=_clean_optional(snapshot.description),
+            status="active",
+            schema_version=1,
+            schema_=[_column_payload(column) for column in schema],
+            rows=rows,
+            fingerprint=fingerprint,
+            row_count=len(rows),
+            byte_count=len(encoded),
+            metadata_=dict(snapshot.metadata),
+            created_by=actor_id,
+            updated_by=actor_id,
+        )
+        db.add(item)
+        db.flush()
+        return _source_dto(item)
+
+    def delete_snapshot(
+        self,
+        session: object,
+        principal: object,
+        *,
+        source_ref: str,
+    ) -> TabularSource:
+        db, api_principal = _context(session, principal, WRITE_SCOPE)
+        item = _source_record(
+            db,
+            tenant_id=api_principal.tenant_id,
+            source_ref=source_ref,
+        )
+        if item is None:
+            raise TabularSourceNotFoundError("Tabular source not found.")
+        item.deleted_at = utcnow()
+        item.status = "retired"
+        item.updated_by = _actor_id(api_principal)
+        db.flush()
+        return _source_dto(item)
+
+
+def parse_csv_snapshot(csv_text: str, *, delimiter: str) -> tuple[Mapping[str, object], ...]:
+    return parse_tabular_csv(
+        csv_text,
+        delimiter=delimiter,
+        max_rows=MAX_SNAPSHOT_ROWS,
+    )
+
+
+def infer_schema(rows: Sequence[Mapping[str, object]]) -> tuple[TabularColumn, ...]:
+    names: list[str] = []
+    for row in rows:
+        for name in row:
+            if name not in names:
+                names.append(name)
+    result: list[TabularColumn] = []
+    for name in names:
+        values = [row.get(name) for row in rows]
+        concrete = [value for value in values if value is not None]
+        data_type = _type_name(concrete[0]) if concrete else "unknown"
+        if any(_type_name(value) != data_type for value in concrete[1:]):
+            data_type = "mixed"
+        result.append(
+            TabularColumn(
+                name=name,
+                data_type=data_type,
+                nullable=len(concrete) != len(values),
+            )
+        )
+    return tuple(result)
+
+
+def snapshot_fingerprint(
+    rows: Sequence[Mapping[str, object]],
+    schema: Sequence[TabularColumn],
+) -> str:
+    payload = {
+        "schema": [_column_payload(column) for column in schema],
+        "rows": [dict(row) for row in rows],
+    }
+    encoded = json.dumps(payload, sort_keys=True, separators=(",", ":"), default=str)
+    return hashlib.sha256(encoded.encode("utf-8")).hexdigest()
+
+
+def _source_record(
+    session: Session,
+    *,
+    tenant_id: str,
+    source_ref: str,
+) -> ConnectorTabularSource | None:
+    source_id = source_ref.removeprefix("snapshot:")
+    if not source_id or source_id == source_ref:
+        return None
+    return session.scalar(
+        select(ConnectorTabularSource).where(
+            ConnectorTabularSource.id == source_id,
+            ConnectorTabularSource.tenant_id == tenant_id,
+            ConnectorTabularSource.deleted_at.is_(None),
+        )
+    )
+
+
+def _source_dto(item: ConnectorTabularSource) -> TabularSource:
+    return TabularSource(
+        ref=f"snapshot:{item.id}",
+        provider=item.provider,
+        source_name=item.source_name,
+        name=item.name,
+        description=item.description,
+        schema=tuple(TabularColumn(**column) for column in item.schema_),
+        schema_version=str(item.schema_version),
+        fingerprint=item.fingerprint,
+        row_count=item.row_count,
+        byte_count=item.byte_count,
+        updated_at=item.updated_at,
+        capabilities=("read", "preview"),
+        metadata=dict(item.metadata_),
+    )
+
+
+def _context(
+    session: object,
+    principal: object,
+    required_scope: str,
+) -> tuple[Session, ApiPrincipal]:
+    if not isinstance(session, Session):
+        raise TypeError("Tabular source providers require a SQLAlchemy session.")
+    if not isinstance(principal, ApiPrincipal):
+        raise TabularSourceAccessError("A tenant API principal is required.")
+    if not (has_scope(principal, required_scope) or has_scope(principal, ADMIN_SCOPE)):
+        raise TabularSourceAccessError(f"Missing scope: {required_scope}")
+    return session, principal
+
+
+def _json_row(row: Mapping[str, object]) -> dict[str, Any]:
+    normalized = {str(key).strip(): value for key, value in row.items()}
+    if not normalized or any(not key for key in normalized):
+        raise TabularSourceValidationError("Every snapshot row needs named columns.")
+    try:
+        json.dumps(normalized, default=_unsupported_json)
+    except (TypeError, ValueError) as exc:
+        raise TabularSourceValidationError(f"Snapshot values must be JSON compatible: {exc}") from exc
+    return json.loads(json.dumps(normalized, default=_unsupported_json))
+
+
+def _unsupported_json(value: object) -> object:
+    if isinstance(value, (datetime, Decimal)):
+        return str(value)
+    raise TypeError(f"{type(value).__name__} is not JSON serializable")
+
+
+def _type_name(value: object) -> str:
+    if isinstance(value, bool):
+        return "boolean"
+    if isinstance(value, int):
+        return "integer"
+    if isinstance(value, (float, Decimal)):
+        return "number"
+    if isinstance(value, str):
+        return "string"
+    if isinstance(value, list):
+        return "array"
+    if isinstance(value, dict):
+        return "object"
+    return type(value).__name__.lower()
+
+
+def _column_payload(column: TabularColumn) -> dict[str, object]:
+    return {
+        "name": column.name,
+        "data_type": column.data_type,
+        "nullable": column.nullable,
+    }
+
+
+def _actor_id(principal: ApiPrincipal) -> str | None:
+    return principal.account_id or principal.membership_id or principal.identity_id
+
+
+def _clean_optional(value: str | None) -> str | None:
+    cleaned = str(value or "").strip()
+    return cleaned or None
+
+
+def _escape_like(value: str) -> str:
+    return value.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
+
+
+__all__ = [
+    "ADMIN_SCOPE",
+    "MAX_READ_ROWS",
+    "MAX_SNAPSHOT_BYTES",
+    "MAX_SNAPSHOT_ROWS",
+    "READ_SCOPE",
+    "SqlTabularSourceProvider",
+    "WRITE_SCOPE",
+    "infer_schema",
+    "parse_csv_snapshot",
+    "snapshot_fingerprint",
+]

src/govoplan_connectors/py.typed

@@ -0,0 +1 @@
+

tests/test_manifest.py

@@ -0,0 +1,32 @@
+from __future__ import annotations
+
+import unittest
+
+from govoplan_core.core.tabular_sources import (
+    CAPABILITY_CONNECTORS_TABULAR_SNAPSHOT_WRITER,
+    CAPABILITY_CONNECTORS_TABULAR_SOURCES,
+)
+from govoplan_connectors.backend.manifest import manifest
+
+
+class ConnectorsManifestTests(unittest.TestCase):
+    def test_manifest_exposes_versioned_tabular_capabilities(self) -> None:
+        self.assertEqual("connectors", manifest.id)
+        self.assertIn("access", manifest.optional_dependencies)
+        self.assertIn(
+            "connectors.tabular_sources",
+            {interface.name for interface in manifest.provides_interfaces},
+        )
+        self.assertIn(
+            CAPABILITY_CONNECTORS_TABULAR_SOURCES,
+            manifest.capability_factories,
+        )
+        self.assertIn(
+            CAPABILITY_CONNECTORS_TABULAR_SNAPSHOT_WRITER,
+            manifest.capability_factories,
+        )
+        self.assertIsNotNone(manifest.migration_spec)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_migrations.py

@@ -0,0 +1,39 @@
+from __future__ import annotations
+
+import tempfile
+import unittest
+from pathlib import Path
+
+from alembic.runtime.migration import MigrationContext
+from sqlalchemy import create_engine, inspect
+
+from govoplan_connectors.backend.manifest import get_manifest
+from govoplan_core.db.migrations import migrate_database
+
+
+class ConnectorsMigrationTests(unittest.TestCase):
+    def test_baseline_creates_connector_tables_and_head(self) -> None:
+        with tempfile.TemporaryDirectory(prefix="govoplan-connectors-migration-") as directory:
+            url = f"sqlite:///{Path(directory) / 'connectors.db'}"
+            migrate_database(
+                database_url=url,
+                enabled_modules=("connectors",),
+                manifest_factories=(get_manifest,),
+            )
+            engine = create_engine(url)
+            try:
+                with engine.connect() as connection:
+                    self.assertIn(
+                        "e6b7c8d9f0a1",
+                        set(MigrationContext.configure(connection).get_current_heads()),
+                    )
+                    self.assertIn(
+                        "connector_tabular_sources",
+                        inspect(connection).get_table_names(),
+                    )
+            finally:
+                engine.dispose()
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_tabular_sources.py

@@ -0,0 +1,179 @@
+from __future__ import annotations
+
+import unittest
+
+from fastapi import HTTPException
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+from govoplan_core.auth import ApiPrincipal
+from govoplan_core.core.access import PrincipalRef
+from govoplan_core.core.tabular_sources import (
+    TabularReadRequest,
+    TabularSnapshotInput,
+    TabularSourceAccessError,
+    TabularSourceValidationError,
+)
+from govoplan_core.db.base import Base
+from govoplan_connectors.backend.db.models import ConnectorTabularSource
+from govoplan_connectors.backend.router import api_create_tabular_snapshot
+from govoplan_connectors.backend.schemas import SnapshotCreateRequest
+from govoplan_connectors.backend.tabular_sources import (
+    READ_SCOPE,
+    WRITE_SCOPE,
+    SqlTabularSourceProvider,
+    parse_csv_snapshot,
+)
+
+
+def principal(
+    tenant_id: str = "tenant-1",
+    *,
+    scopes: tuple[str, ...] = (READ_SCOPE, WRITE_SCOPE),
+) -> ApiPrincipal:
+    return ApiPrincipal(
+        principal=PrincipalRef(
+            account_id="account-1",
+            membership_id="membership-1",
+            tenant_id=tenant_id,
+            scopes=frozenset(scopes),
+        ),
+        account=object(),
+        user=object(),
+    )
+
+
+class ConnectorsTabularSourceTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.engine = create_engine("sqlite:///:memory:")
+        Base.metadata.create_all(self.engine, tables=[ConnectorTabularSource.__table__])
+        self.Session = sessionmaker(bind=self.engine)
+        self.session = self.Session()
+        self.provider = SqlTabularSourceProvider()
+
+    def tearDown(self) -> None:
+        self.session.close()
+        Base.metadata.drop_all(self.engine, tables=[ConnectorTabularSource.__table__])
+        self.engine.dispose()
+
+    def test_snapshot_round_trip_preserves_schema_fingerprint_and_bounds(self) -> None:
+        created = self.provider.create_snapshot(
+            self.session,
+            principal(),
+            snapshot=TabularSnapshotInput(
+                name="Monthly cases",
+                source_name="monthly_cases_2026_07",
+                rows=(
+                    {"case_id": "A-1", "amount": 12, "active": True},
+                    {"case_id": "A-2", "amount": None, "active": False},
+                ),
+            ),
+        )
+        self.session.commit()
+
+        listed = self.provider.list_sources(self.session, principal())
+        preview = self.provider.read_source(
+            self.session,
+            principal(),
+            request=TabularReadRequest(
+                source_ref=created.ref,
+                limit=1,
+                expected_fingerprint=created.fingerprint,
+            ),
+        )
+
+        self.assertEqual((created.ref,), tuple(source.ref for source in listed))
+        self.assertEqual(
+            ["case_id", "amount", "active"],
+            [column.name for column in created.schema],
+        )
+        self.assertEqual(2, preview.total_rows)
+        self.assertEqual(1, len(preview.rows))
+        self.assertTrue(preview.truncated)
+        self.assertEqual(created.fingerprint, preview.source.fingerprint)
+
+    def test_tenant_and_scope_isolation_are_enforced(self) -> None:
+        created = self.provider.create_snapshot(
+            self.session,
+            principal(),
+            snapshot=TabularSnapshotInput(
+                name="Private",
+                source_name="private_source",
+                rows=({"id": 1},),
+            ),
+        )
+        self.session.commit()
+
+        self.assertEqual((), self.provider.list_sources(self.session, principal("tenant-2")))
+        self.assertIsNone(
+            self.provider.get_source(
+                self.session,
+                principal("tenant-2"),
+                source_ref=created.ref,
+            )
+        )
+        with self.assertRaises(TabularSourceAccessError):
+            self.provider.list_sources(
+                self.session,
+                principal(scopes=()),
+            )
+
+    def test_duplicate_source_name_and_stale_fingerprint_are_rejected(self) -> None:
+        snapshot = TabularSnapshotInput(
+            name="Cases",
+            source_name="cases",
+            rows=({"id": 1},),
+        )
+        created = self.provider.create_snapshot(self.session, principal(), snapshot=snapshot)
+        self.session.commit()
+
+        with self.assertRaises(TabularSourceValidationError):
+            self.provider.create_snapshot(self.session, principal(), snapshot=snapshot)
+        with self.assertRaises(TabularSourceValidationError):
+            self.provider.read_source(
+                self.session,
+                principal(),
+                request=TabularReadRequest(
+                    source_ref=created.ref,
+                    expected_fingerprint="stale",
+                ),
+            )
+
+    def test_csv_parser_infers_scalar_values_and_rejects_duplicate_headers(self) -> None:
+        rows = parse_csv_snapshot(
+            "\ufeffid;amount;active;note\n0012;12.5;true;\n2;7;false;ok\n",
+            delimiter=";",
+        )
+
+        self.assertEqual(
+            (
+                {"id": "0012", "amount": 12.5, "active": True, "note": None},
+                {"id": 2, "amount": 7, "active": False, "note": "ok"},
+            ),
+            rows,
+        )
+        with self.assertRaises(TabularSourceValidationError):
+            parse_csv_snapshot("id,id\n1,2\n", delimiter=",")
+        with self.assertRaises(TabularSourceValidationError):
+            parse_csv_snapshot("id,name\n1,Ada,extra\n", delimiter=",")
+
+    def test_malformed_csv_api_request_is_reported_as_validation_error(self) -> None:
+        payload = SnapshotCreateRequest(
+            name="Malformed",
+            source_name="malformed",
+            format="csv",
+            csv_text="id,name\n1,Ada,extra\n",
+        )
+
+        with self.assertRaises(HTTPException) as raised:
+            api_create_tabular_snapshot(
+                payload,
+                session=self.session,
+                principal=principal(),
+            )
+
+        self.assertEqual(422, raised.exception.status_code)
+
+
+if __name__ == "__main__":
+    unittest.main()