From ba5ccea5b09fd2623bbca054b71dce21f1b4f56f Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Tue, 28 Jul 2026 11:13:22 +0200 Subject: [PATCH] Implement governed tabular source snapshots --- README.md | 22 +- pyproject.toml | 22 ++ src/govoplan_connectors/__init__.py | 3 + src/govoplan_connectors/backend/__init__.py | 1 + .../backend/db/__init__.py | 3 + src/govoplan_connectors/backend/db/models.py | 44 +++ src/govoplan_connectors/backend/manifest.py | 190 +++++++++ .../backend/migrations/__init__.py | 1 + .../backend/migrations/versions/__init__.py | 1 + .../e6b7c8d9f0a1_v0114_connectors_baseline.py | 141 +++++++ src/govoplan_connectors/backend/router.py | 211 ++++++++++ src/govoplan_connectors/backend/schemas.py | 79 ++++ .../backend/tabular_sources.py | 365 ++++++++++++++++++ src/govoplan_connectors/py.typed | 1 + tests/test_manifest.py | 32 ++ tests/test_migrations.py | 39 ++ tests/test_tabular_sources.py | 179 +++++++++ 17 files changed, 1332 insertions(+), 2 deletions(-) create mode 100644 pyproject.toml create mode 100644 src/govoplan_connectors/__init__.py create mode 100644 src/govoplan_connectors/backend/__init__.py create mode 100644 src/govoplan_connectors/backend/db/__init__.py create mode 100644 src/govoplan_connectors/backend/db/models.py create mode 100644 src/govoplan_connectors/backend/manifest.py create mode 100644 src/govoplan_connectors/backend/migrations/__init__.py create mode 100644 src/govoplan_connectors/backend/migrations/versions/__init__.py create mode 100644 src/govoplan_connectors/backend/migrations/versions/e6b7c8d9f0a1_v0114_connectors_baseline.py create mode 100644 src/govoplan_connectors/backend/router.py create mode 100644 src/govoplan_connectors/backend/schemas.py create mode 100644 src/govoplan_connectors/backend/tabular_sources.py create mode 100644 src/govoplan_connectors/py.typed create mode 100644 tests/test_manifest.py create mode 100644 tests/test_migrations.py create mode 100644 tests/test_tabular_sources.py diff --git a/README.md b/README.md index e2cbbcd..ffd4885 100644 --- a/README.md +++ b/README.md @@ -4,14 +4,32 @@ **Repository type:** connector (connector-hub). -`govoplan-connectors` will own integration catalogues and generic external -system connection patterns for GovOPlaN. +`govoplan-connectors` owns integration catalogues and generic external system +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) diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..a21cbbc --- /dev/null +++ b/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" diff --git a/src/govoplan_connectors/__init__.py b/src/govoplan_connectors/__init__.py new file mode 100644 index 0000000..89108e4 --- /dev/null +++ b/src/govoplan_connectors/__init__.py @@ -0,0 +1,3 @@ +"""GovOPlaN Connectors module.""" + +__all__: list[str] = [] diff --git a/src/govoplan_connectors/backend/__init__.py b/src/govoplan_connectors/backend/__init__.py new file mode 100644 index 0000000..2a925ad --- /dev/null +++ b/src/govoplan_connectors/backend/__init__.py @@ -0,0 +1 @@ +"""Connector backend package.""" diff --git a/src/govoplan_connectors/backend/db/__init__.py b/src/govoplan_connectors/backend/db/__init__.py new file mode 100644 index 0000000..0f9dc45 --- /dev/null +++ b/src/govoplan_connectors/backend/db/__init__.py @@ -0,0 +1,3 @@ +from govoplan_connectors.backend.db.models import ConnectorTabularSource + +__all__ = ["ConnectorTabularSource"] diff --git a/src/govoplan_connectors/backend/db/models.py b/src/govoplan_connectors/backend/db/models.py new file mode 100644 index 0000000..fde304f --- /dev/null +++ b/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"] diff --git a/src/govoplan_connectors/backend/manifest.py b/src/govoplan_connectors/backend/manifest.py new file mode 100644 index 0000000..54dc5b2 --- /dev/null +++ b/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", +] diff --git a/src/govoplan_connectors/backend/migrations/__init__.py b/src/govoplan_connectors/backend/migrations/__init__.py new file mode 100644 index 0000000..7e57e86 --- /dev/null +++ b/src/govoplan_connectors/backend/migrations/__init__.py @@ -0,0 +1 @@ +"""Connectors migrations.""" diff --git a/src/govoplan_connectors/backend/migrations/versions/__init__.py b/src/govoplan_connectors/backend/migrations/versions/__init__.py new file mode 100644 index 0000000..64e4b7c --- /dev/null +++ b/src/govoplan_connectors/backend/migrations/versions/__init__.py @@ -0,0 +1 @@ +"""Connectors migration revisions.""" diff --git a/src/govoplan_connectors/backend/migrations/versions/e6b7c8d9f0a1_v0114_connectors_baseline.py b/src/govoplan_connectors/backend/migrations/versions/e6b7c8d9f0a1_v0114_connectors_baseline.py new file mode 100644 index 0000000..1f7c8df --- /dev/null +++ b/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") diff --git a/src/govoplan_connectors/backend/router.py b/src/govoplan_connectors/backend/router.py new file mode 100644 index 0000000..604c6d9 --- /dev/null +++ b/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"] diff --git a/src/govoplan_connectors/backend/schemas.py b/src/govoplan_connectors/backend/schemas.py new file mode 100644 index 0000000..986d3c8 --- /dev/null +++ b/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", +] diff --git a/src/govoplan_connectors/backend/tabular_sources.py b/src/govoplan_connectors/backend/tabular_sources.py new file mode 100644 index 0000000..a515542 --- /dev/null +++ b/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", +] diff --git a/src/govoplan_connectors/py.typed b/src/govoplan_connectors/py.typed new file mode 100644 index 0000000..8b13789 --- /dev/null +++ b/src/govoplan_connectors/py.typed @@ -0,0 +1 @@ + diff --git a/tests/test_manifest.py b/tests/test_manifest.py new file mode 100644 index 0000000..572421d --- /dev/null +++ b/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() diff --git a/tests/test_migrations.py b/tests/test_migrations.py new file mode 100644 index 0000000..f223fbd --- /dev/null +++ b/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() diff --git a/tests/test_tabular_sources.py b/tests/test_tabular_sources.py new file mode 100644 index 0000000..2280f5e --- /dev/null +++ b/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()