diff --git a/README.md b/README.md index 856b21c..707fe5f 100644 --- a/README.md +++ b/README.md @@ -17,7 +17,9 @@ or reporting behavior. The first executable connector capability provides tenant-isolated tabular origins. Operators can import bounded JSON or CSV snapshots, inspect inferred schemas, and expose immutable source references and content fingerprints -through `connectors.datasource_origins@0.1.0`. +through `connectors.datasource_origins@0.1.0`. Preview reads enforce provider +ceilings for rows, serialized bytes, and elapsed time and report the effective +limits and any truncation as structured diagnostics. Connectors owns acquisition, connection profiles, credentials, discovery, and provider health. `govoplan-datasources` registers an origin as a governed live @@ -25,6 +27,12 @@ or cached datasource and owns staging, materializations, frozen states, and consumer access. Dataflow consumes that Datasources contract and never imports connector implementations or stores connector credentials. +Each origin declares whether it is live, cached, file-backed, or static, its +structured health state, and which projection, filter, aggregation, sorting, +and pagination operations it can push down. The immutable snapshot provider +currently supports projection and pagination only; consumers must keep other +operations in Dataflow rather than assuming transport-side execution. + Database, REST/HTTP, directory, managed-file, and warehouse providers can implement the same origin contract without changing Datasources or Dataflow. diff --git a/src/govoplan_connectors/backend/datasource_origins.py b/src/govoplan_connectors/backend/datasource_origins.py index 2b6a970..eeda54c 100644 --- a/src/govoplan_connectors/backend/datasource_origins.py +++ b/src/govoplan_connectors/backend/datasource_origins.py @@ -7,6 +7,7 @@ from govoplan_core.core.datasources import ( DatasourceOrigin, DatasourceOriginReadRequest, DatasourceOriginReadResult, + DatasourceUnavailableError, DatasourceValidationError, ) from govoplan_core.core.tabular_sources import ( @@ -15,6 +16,7 @@ from govoplan_core.core.tabular_sources import ( TabularSourceAccessError, TabularSourceError, TabularSourceNotFoundError, + TabularSourceUnavailableError, TabularSourceValidationError, ) from govoplan_connectors.backend.tabular_sources import SqlTabularSourceProvider @@ -79,6 +81,8 @@ class ConnectorDatasourceOriginProvider: offset=request.offset, columns=request.columns, expected_fingerprint=request.expected_fingerprint, + max_bytes=request.max_bytes, + timeout_ms=request.timeout_ms, ), ) except TabularSourceError as exc: @@ -88,6 +92,12 @@ class ConnectorDatasourceOriginProvider: rows=result.rows, total_rows=result.total_rows, truncated=result.truncated, + returned_bytes=result.returned_bytes, + elapsed_ms=result.elapsed_ms, + effective_row_limit=result.effective_row_limit, + effective_byte_limit=result.effective_byte_limit, + effective_timeout_ms=result.effective_timeout_ms, + diagnostics=result.diagnostics, ) @@ -116,6 +126,9 @@ def _origin(source: TabularSource) -> DatasourceOrigin: updated_at=source.updated_at, capabilities=source.capabilities, metadata=dict(source.metadata), + source_mode=source.source_mode, + pushdown=source.pushdown, + health=source.health, ) @@ -124,6 +137,8 @@ def _datasource_error(exc: TabularSourceError): return DatasourceAccessError(str(exc)) if isinstance(exc, TabularSourceNotFoundError): return DatasourceNotFoundError(str(exc)) + if isinstance(exc, TabularSourceUnavailableError): + return DatasourceUnavailableError(str(exc)) if isinstance(exc, TabularSourceValidationError): return DatasourceValidationError(str(exc)) return DatasourceValidationError(str(exc)) diff --git a/src/govoplan_connectors/backend/manifest.py b/src/govoplan_connectors/backend/manifest.py index cf5a8af..9d307b8 100644 --- a/src/govoplan_connectors/backend/manifest.py +++ b/src/govoplan_connectors/backend/manifest.py @@ -453,7 +453,9 @@ manifest = ModuleManifest( "Connectors owns endpoint discovery, authentication hand-off, transport limits, retries, and protocol health. " "Domain modules own field mapping, validation, reconciliation, and record mutation. The shared Core runtime " "contract reports redacted effects and diagnostics with source revisions, fingerprints, and immutable input hashes. " - "A commit must reject stale, truncated, conflicting, or error-bearing previews, and credentials never appear in URLs or samples." + "Tabular previews enforce effective row, serialized-byte, and elapsed-time ceilings and report limit truncation " + "as structured diagnostics. A commit must reject stale, truncated, conflicting, or error-bearing previews, and " + "credentials never appear in URLs or samples." ), layer="available", documentation_types=("admin", "user"), @@ -468,8 +470,11 @@ manifest = ModuleManifest( 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 and exposes them as Datasource " + "references and expected fingerprints. Each source declares its live, " + "cached, file-backed, or static mode, structured health, and supported " + "projection, filter, aggregation, sorting, and pagination pushdown. The " + "first executable provider imports immutable JSON or CSV snapshots, " + "supports projection and pagination, and exposes them as Datasource " "origins. Database and API providers can implement the same origin " "contract without changing Datasources or Dataflow." ), diff --git a/src/govoplan_connectors/backend/router.py b/src/govoplan_connectors/backend/router.py index c0d1af8..796395b 100644 --- a/src/govoplan_connectors/backend/router.py +++ b/src/govoplan_connectors/backend/router.py @@ -16,6 +16,7 @@ from govoplan_core.core.tabular_sources import ( TabularSourceAccessError, TabularSourceError, TabularSourceNotFoundError, + TabularSourceUnavailableError, ) from govoplan_core.core.feeds import ( FeedCapabilityError, @@ -38,6 +39,9 @@ from govoplan_connectors.backend.schemas import ( SanctionsSourceResponse, SnapshotCreateRequest, TabularColumnResponse, + TabularHealthResponse, + TabularPreviewDiagnosticResponse, + TabularPushdownResponse, TabularSourceDeleteResponse, TabularSourceListResponse, TabularSourcePreviewResponse, @@ -85,6 +89,11 @@ def _http_error(exc: TabularSourceError) -> HTTPException: 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)) + if isinstance(exc, TabularSourceUnavailableError): + return HTTPException( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, + detail=str(exc), + ) return HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc)) @@ -377,6 +386,8 @@ def api_preview_tabular_source( source_id: str, limit: int = Query(default=100, ge=1, le=500), offset: int = Query(default=0, ge=0), + max_bytes: int = Query(default=1_000_000, ge=2, le=5_000_000), + timeout_ms: int = Query(default=2_000, ge=1, le=10_000), session: Session = Depends(get_session), principal: ApiPrincipal = Depends(get_api_principal), ) -> TabularSourcePreviewResponse: @@ -389,6 +400,8 @@ def api_preview_tabular_source( source_ref=f"snapshot:{source_id}", limit=limit, offset=offset, + max_bytes=max_bytes, + timeout_ms=timeout_ms, ), ) except TabularSourceError as exc: @@ -398,6 +411,20 @@ def api_preview_tabular_source( rows=[dict(row) for row in result.rows], total_rows=result.total_rows, truncated=result.truncated, + returned_bytes=result.returned_bytes, + elapsed_ms=result.elapsed_ms, + effective_row_limit=result.effective_row_limit, + effective_byte_limit=result.effective_byte_limit, + effective_timeout_ms=result.effective_timeout_ms, + diagnostics=[ + TabularPreviewDiagnosticResponse( + severity=item.severity, + code=item.code, + message=item.message, + details=dict(item.details), + ) + for item in result.diagnostics + ], ) @@ -604,6 +631,25 @@ def _source_response(source: TabularSource) -> TabularSourceResponse: updated_at=source.updated_at.isoformat() if source.updated_at else None, capabilities=list(source.capabilities), metadata=dict(source.metadata), + source_mode=source.source_mode, + pushdown=TabularPushdownResponse( + projections=source.pushdown.projections, + pagination=source.pushdown.pagination, + filters=list(source.pushdown.filters), + aggregations=list(source.pushdown.aggregations), + sorting=list(source.pushdown.sorting), + ), + health=TabularHealthResponse( + status=source.health.status, + code=source.health.code, + summary=source.health.summary, + checked_at=( + source.health.checked_at.isoformat() + if source.health.checked_at + else None + ), + details=dict(source.health.details), + ), ) diff --git a/src/govoplan_connectors/backend/schemas.py b/src/govoplan_connectors/backend/schemas.py index c22676a..6de5947 100644 --- a/src/govoplan_connectors/backend/schemas.py +++ b/src/govoplan_connectors/backend/schemas.py @@ -100,6 +100,29 @@ class TabularColumnResponse(BaseModel): nullable: bool +class TabularPushdownResponse(BaseModel): + projections: bool + pagination: bool + filters: list[str] + aggregations: list[str] + sorting: list[str] + + +class TabularHealthResponse(BaseModel): + status: Literal["healthy", "warning", "error", "unknown"] + code: str + summary: str + checked_at: str | None + details: dict[str, Any] + + +class TabularPreviewDiagnosticResponse(BaseModel): + severity: Literal["info", "warning", "error"] + code: str + message: str + details: dict[str, Any] + + class TabularSourceResponse(BaseModel): ref: str provider: str @@ -114,6 +137,9 @@ class TabularSourceResponse(BaseModel): updated_at: str | None capabilities: list[str] metadata: dict[str, Any] + source_mode: Literal["live", "cached", "file_backed", "static"] + pushdown: TabularPushdownResponse + health: TabularHealthResponse class TabularSourceListResponse(BaseModel): @@ -125,6 +151,12 @@ class TabularSourcePreviewResponse(BaseModel): rows: list[dict[str, Any]] total_rows: int truncated: bool + returned_bytes: int + elapsed_ms: int + effective_row_limit: int + effective_byte_limit: int + effective_timeout_ms: int + diagnostics: list[TabularPreviewDiagnosticResponse] class TabularSourceDeleteResponse(BaseModel): @@ -216,6 +248,9 @@ __all__ = [ "SanctionsSourceListResponse", "SanctionsSourceResponse", "TabularColumnResponse", + "TabularHealthResponse", + "TabularPreviewDiagnosticResponse", + "TabularPushdownResponse", "TabularSourceDeleteResponse", "TabularSourceListResponse", "TabularSourcePreviewResponse", diff --git a/src/govoplan_connectors/backend/tabular_sources.py b/src/govoplan_connectors/backend/tabular_sources.py index a8842c5..07b9bce 100644 --- a/src/govoplan_connectors/backend/tabular_sources.py +++ b/src/govoplan_connectors/backend/tabular_sources.py @@ -2,10 +2,11 @@ from __future__ import annotations import hashlib import json +import time from collections.abc import Mapping, Sequence from datetime import datetime from decimal import Decimal -from typing import Any +from typing import Any, Callable from sqlalchemy import or_, select from sqlalchemy.orm import Session @@ -13,12 +14,16 @@ from sqlalchemy.orm import Session from govoplan_core.auth import ApiPrincipal, has_scope from govoplan_core.core.tabular_sources import ( TabularColumn, + TabularPreviewDiagnostic, + TabularPushdown, TabularReadRequest, TabularReadResult, TabularSnapshotInput, TabularSource, TabularSourceAccessError, TabularSourceNotFoundError, + TabularSourceHealth, + TabularSourceUnavailableError, TabularSourceValidationError, parse_tabular_csv, ) @@ -32,9 +37,14 @@ ADMIN_SCOPE = "connectors:source:admin" MAX_SNAPSHOT_ROWS = 10_000 MAX_SNAPSHOT_BYTES = 5_000_000 MAX_READ_ROWS = 500 +MAX_READ_BYTES = 1_000_000 +MAX_READ_TIMEOUT_MS = 2_000 class SqlTabularSourceProvider: + def __init__(self, *, clock: Callable[[], float] = time.monotonic) -> None: + self._clock = clock + def list_sources( self, session: object, @@ -99,8 +109,39 @@ class SqlTabularSourceProvider: "The source fingerprint changed; refresh the source node before running it." ) + started = self._clock() limit = max(1, min(int(request.limit), MAX_READ_ROWS)) + byte_limit = max(2, min(int(request.max_bytes), MAX_READ_BYTES)) + timeout_ms = max(1, min(int(request.timeout_ms), MAX_READ_TIMEOUT_MS)) offset = max(0, int(request.offset)) + diagnostics: list[TabularPreviewDiagnostic] = [] + if limit != request.limit: + diagnostics.append( + _preview_diagnostic( + "preview.row_limit_tightened", + "The provider tightened the requested row limit.", + requested=request.limit, + effective=limit, + ) + ) + if byte_limit != request.max_bytes: + diagnostics.append( + _preview_diagnostic( + "preview.byte_limit_tightened", + "The provider tightened the requested byte limit.", + requested=request.max_bytes, + effective=byte_limit, + ) + ) + if timeout_ms != request.timeout_ms: + diagnostics.append( + _preview_diagnostic( + "preview.timeout_tightened", + "The provider tightened the requested time limit.", + requested=request.timeout_ms, + effective=timeout_ms, + ) + ) 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] @@ -108,20 +149,70 @@ class SqlTabularSourceProvider: raise TabularSourceValidationError( f"Unknown source columns: {', '.join(unknown_columns)}" ) - window = item.rows[offset : offset + limit] - rows = tuple( - { + rows: list[dict[str, object]] = [] + returned_bytes = 2 + stopped_for = "" + for row in item.rows[offset:]: + if len(rows) >= limit: + stopped_for = "rows" + break + elapsed_ms = int(max(0.0, self._clock() - started) * 1_000) + if elapsed_ms >= timeout_ms: + if not rows: + raise TabularSourceUnavailableError( + "Tabular source preview exceeded its time budget." + ) + stopped_for = "time" + break + selected = { key: value for key, value in row.items() if not selected_columns or key in selected_columns } - for row in window - ) + row_bytes = len( + json.dumps( + selected, + sort_keys=True, + separators=(",", ":"), + default=str, + ).encode("utf-8") + ) + additional_bytes = row_bytes + (1 if rows else 0) + if returned_bytes + additional_bytes > byte_limit: + if not rows: + raise TabularSourceValidationError( + "A single source row exceeds the preview byte limit." + ) + stopped_for = "bytes" + break + rows.append(selected) + returned_bytes += additional_bytes + elapsed_ms = int(max(0.0, self._clock() - started) * 1_000) + if stopped_for: + labels = { + "rows": ("preview.row_limit_reached", "row"), + "bytes": ("preview.byte_limit_reached", "byte"), + "time": ("preview.timeout_reached", "time"), + } + code, label = labels[stopped_for] + diagnostics.append( + TabularPreviewDiagnostic( + severity="warning", + code=code, + message=f"The preview stopped at its effective {label} limit.", + ) + ) return TabularReadResult( source=_source_dto(item), - rows=rows, + rows=tuple(rows), total_rows=item.row_count, truncated=offset + len(rows) < item.row_count, + returned_bytes=returned_bytes, + elapsed_ms=elapsed_ms, + effective_row_limit=limit, + effective_byte_limit=byte_limit, + effective_timeout_ms=timeout_ms, + diagnostics=tuple(diagnostics), ) def create_snapshot( @@ -283,6 +374,33 @@ def _source_dto(item: ConnectorTabularSource) -> TabularSource: updated_at=item.updated_at, capabilities=("read", "preview"), metadata=dict(item.metadata_), + source_mode="cached", + pushdown=TabularPushdown( + projections=True, + pagination=True, + ), + health=TabularSourceHealth( + status="healthy", + code="snapshot.ready", + summary="The immutable connector snapshot is ready.", + checked_at=item.updated_at, + details={"immutable": True}, + ), + ) + + +def _preview_diagnostic( + code: str, + message: str, + *, + requested: int, + effective: int, +) -> TabularPreviewDiagnostic: + return TabularPreviewDiagnostic( + severity="info", + code=code, + message=message, + details={"requested": requested, "effective": effective}, ) @@ -367,6 +485,8 @@ def _escape_like(value: str) -> str: __all__ = [ "ADMIN_SCOPE", "MAX_READ_ROWS", + "MAX_READ_BYTES", + "MAX_READ_TIMEOUT_MS", "MAX_SNAPSHOT_BYTES", "MAX_SNAPSHOT_ROWS", "READ_SCOPE", diff --git a/tests/test_datasource_origins.py b/tests/test_datasource_origins.py index c61be8a..ea3862e 100644 --- a/tests/test_datasource_origins.py +++ b/tests/test_datasource_origins.py @@ -81,6 +81,11 @@ class ConnectorDatasourceOriginTests(unittest.TestCase): self.assertEqual((self.source.ref,), tuple(item.ref for item in origins)) self.assertEqual(("live", "cached"), origins[0].supported_modes) self.assertEqual(({"id": 1, "name": "Ada"},), result.rows) + self.assertEqual("cached", origins[0].source_mode) + self.assertTrue(origins[0].pushdown.projections) + self.assertEqual("healthy", origins[0].health.status) + self.assertGreater(result.returned_bytes, 2) + self.assertEqual(1_000_000, result.effective_byte_limit) if __name__ == "__main__": diff --git a/tests/test_tabular_sources.py b/tests/test_tabular_sources.py index 2280f5e..f5725eb 100644 --- a/tests/test_tabular_sources.py +++ b/tests/test_tabular_sources.py @@ -12,6 +12,7 @@ from govoplan_core.core.tabular_sources import ( TabularReadRequest, TabularSnapshotInput, TabularSourceAccessError, + TabularSourceUnavailableError, TabularSourceValidationError, ) from govoplan_core.db.base import Base @@ -91,6 +92,94 @@ class ConnectorsTabularSourceTests(unittest.TestCase): self.assertEqual(1, len(preview.rows)) self.assertTrue(preview.truncated) self.assertEqual(created.fingerprint, preview.source.fingerprint) + self.assertEqual("cached", preview.source.source_mode) + self.assertTrue(preview.source.pushdown.projections) + self.assertTrue(preview.source.pushdown.pagination) + self.assertEqual("healthy", preview.source.health.status) + self.assertGreater(preview.returned_bytes, 2) + self.assertEqual(1, preview.effective_row_limit) + self.assertEqual("preview.row_limit_reached", preview.diagnostics[0].code) + + def test_preview_enforces_byte_time_and_provider_ceiling_budgets(self) -> None: + created = self.provider.create_snapshot( + self.session, + principal(), + snapshot=TabularSnapshotInput( + name="Bounded", + source_name="bounded", + rows=( + {"id": 1, "value": "first"}, + {"id": 2, "value": "second"}, + ), + ), + ) + self.session.commit() + + bounded = self.provider.read_source( + self.session, + principal(), + request=TabularReadRequest( + source_ref=created.ref, + limit=500, + max_bytes=35, + timeout_ms=2_000, + ), + ) + self.assertEqual(1, len(bounded.rows)) + self.assertTrue(bounded.truncated) + self.assertEqual( + "preview.byte_limit_reached", + bounded.diagnostics[-1].code, + ) + with self.assertRaisesRegex( + TabularSourceValidationError, + "single source row exceeds", + ): + self.provider.read_source( + self.session, + principal(), + request=TabularReadRequest( + source_ref=created.ref, + max_bytes=2, + ), + ) + + tightened = self.provider.read_source( + self.session, + principal(), + request=TabularReadRequest( + source_ref=created.ref, + limit=5_000, + max_bytes=5_000_000, + timeout_ms=10_000, + ), + ) + self.assertEqual(500, tightened.effective_row_limit) + self.assertEqual(1_000_000, tightened.effective_byte_limit) + self.assertEqual(2_000, tightened.effective_timeout_ms) + self.assertEqual( + { + "preview.row_limit_tightened", + "preview.byte_limit_tightened", + "preview.timeout_tightened", + }, + {item.code for item in tightened.diagnostics}, + ) + + times = iter((0.0, 0.01)) + timeout_provider = SqlTabularSourceProvider(clock=lambda: next(times)) + with self.assertRaisesRegex( + TabularSourceUnavailableError, + "time budget", + ): + timeout_provider.read_source( + self.session, + principal(), + request=TabularReadRequest( + source_ref=created.ref, + timeout_ms=1, + ), + ) def test_tenant_and_scope_isolation_are_enforced(self) -> None: created = self.provider.create_snapshot(