diff --git a/README.md b/README.md index 97400b2..7edb60f 100644 --- a/README.md +++ b/README.md @@ -14,12 +14,14 @@ or reporting behavior. ## Executable First Slice -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`. Preview reads enforce provider -ceilings for rows, serialized bytes, and elapsed time and report the effective -limits and any truncation as structured diagnostics. +The executable connector capability provides tenant-isolated tabular origins. +Operators can import bounded JSON or CSV snapshots, bind an exact managed Files +CSV/XLSX version, or discover a table through an active governed PostgreSQL +configuration. Every origin exposes a reviewed schema, opaque reference, and +content/discovery fingerprint 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 @@ -29,12 +31,28 @@ 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. +and pagination operations it can push down. The snapshot, managed-file, and +PostgreSQL providers currently support 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. +Managed-file sources are authorized and opened through +`files.tabular_content@1.0.0`; Files remains authoritative for ownership, +shares, download permission, exact versions, integrity, quarantine, encryption, +retention, and legal holds. CSV must be UTF-8. XLSX input is protected by +compressed-entry, expanded-byte, compression-ratio, row, and column limits. +A newer current version is reported but never silently replaces the pinned +version. + +The PostgreSQL adapter accepts only an active governed connector configuration +whose secret-free endpoint uses the PostgreSQL driver. Authentication is +resolved from a tenant/scope/module/server-restricted Core credential envelope. +The adapter reflects a simple schema/table identifier, uses read-only +transactions and a statement timeout, and blocks configuration, credential, or +schema drift until an operator refreshes and reviews the source. Secrets are +never copied into source metadata or diagnostics. Other database, REST/HTTP, +directory, and warehouse providers can implement the same origin contract +without changing Datasources or Dataflow. Governed sanctions and feed snapshot acquisitions use Core recovery operations. The source revision/cursor, redacted dry-run decision, canonical request digest, diff --git a/docs/CONNECTOR_SOURCE_LIFECYCLE.md b/docs/CONNECTOR_SOURCE_LIFECYCLE.md index 509ad02..f7f70d9 100644 --- a/docs/CONNECTOR_SOURCE_LIFECYCLE.md +++ b/docs/CONNECTOR_SOURCE_LIFECYCLE.md @@ -112,6 +112,30 @@ this lifecycle when a connector publishes status. 9. Store external references with source system, object type, object id, version or ETag, and last-seen timestamp. +## Tabular source discovery and refresh + +Immutable JSON/CSV snapshots, exact managed Files versions, and live PostgreSQL +tables share the same catalogue and bounded-preview contract. A managed-file +origin stores only the Files asset id, exact immutable version id, checksum, +parser settings, reviewed schema, and discovery fingerprint. Files re-authorizes +the current principal and verifies storage integrity and any encryption envelope +on every preview. A newer current version produces a warning; only an explicit +source refresh changes the pinned version and increments the discovery revision. + +A PostgreSQL origin references an active governed connector configuration. Its +endpoint must contain no credentials. Connectors resolves the referenced Core +credential envelope for the current tenant, scope, module, and server, opens a +read-only connection, reflects a simple schema/table identifier, and records the +configuration hash/revision, credential revision, schema, and discovery +fingerprint. Configuration, credential, or schema drift blocks preview until an +explicit refresh. Missing Files capability, revoked file access, quarantine, +oversized or malformed content, inactive/stale credentials, unreachable SQL, +and timeout failures produce sanitized unavailable/validation diagnostics. + +All three current providers declare projection and pagination pushdown only. +Filters, aggregations, and sorting remain in Dataflow until an adapter explicitly +declares and tests those operations. + ## Publish Flow 1. Domain module requests publish through a core-mediated connector capability. diff --git a/pyproject.toml b/pyproject.toml index e552511..e4eacef 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "govoplan-connectors" -version = "0.1.19" +version = "0.1.20" description = "Governed connector catalogue and tabular source capabilities for GovOPlaN." readme = "README.md" requires-python = ">=3.12" @@ -12,7 +12,8 @@ license = "AGPL-3.0-or-later" authors = [{ name = "GovOPlaN" }] dependencies = [ "defusedxml>=0.7,<1", - "govoplan-core>=0.1.18", + "govoplan-core>=0.1.19", + "openpyxl>=3.1.5,<4", ] [tool.setuptools.packages.find] diff --git a/src/govoplan_connectors/backend/datasource_origins.py b/src/govoplan_connectors/backend/datasource_origins.py index eeda54c..70d9012 100644 --- a/src/govoplan_connectors/backend/datasource_origins.py +++ b/src/govoplan_connectors/backend/datasource_origins.py @@ -102,12 +102,16 @@ class ConnectorDatasourceOriginProvider: def _origin(source: TabularSource) -> DatasourceOrigin: + kind = { + "managed_file": "file", + "postgresql": "database", + }.get(source.provider, "upload") return DatasourceOrigin( ref=source.ref, source_name=source.source_name, name=source.name, description=source.description, - kind="upload", + kind=kind, shape="tabular", supported_modes=("live", "cached"), provider=f"connectors.{source.provider}", diff --git a/src/govoplan_connectors/backend/manifest.py b/src/govoplan_connectors/backend/manifest.py index d5d8ac4..dc53a9b 100644 --- a/src/govoplan_connectors/backend/manifest.py +++ b/src/govoplan_connectors/backend/manifest.py @@ -18,6 +18,7 @@ from govoplan_core.core.modules import ( FrontendModule, MigrationSpec, ModuleInterfaceProvider, + ModuleInterfaceRequirement, ModuleManifest, PermissionDefinition, RoleTemplate, @@ -36,6 +37,7 @@ from govoplan_core.core.tabular_sources import ( CAPABILITY_CONNECTORS_TABULAR_SNAPSHOT_WRITER, CAPABILITY_CONNECTORS_TABULAR_SOURCES, ) +from govoplan_core.core.files import CAPABILITY_FILES_TABULAR_CONTENT from govoplan_core.core.sanctions import ( CAPABILITY_CONNECTORS_SANCTIONS_SNAPSHOTS, ) @@ -81,7 +83,7 @@ from govoplan_connectors.backend.provider_state import ( MODULE_ID = "connectors" -MODULE_VERSION = "0.1.19" +MODULE_VERSION = "0.1.20" TABULAR_SOURCE_INTERFACE_VERSION = "0.1.0" DATASOURCE_ORIGIN_INTERFACE_VERSION = "0.1.0" SANCTIONS_SNAPSHOT_INTERFACE_VERSION = "1.0.0" @@ -98,6 +100,16 @@ ARCHITECTURE = ModuleArchitectureDeclaration( reference="tests/test_tabular_sources.py", summary="Exercises tenant-safe immutable tabular snapshots and bounded reads.", ), + ModuleMaturityEvidence( + kind="test", + reference="tests/test_tabular_origin_provider.py", + summary="Exercises exact managed-file versions, reviewed refresh, live SQL projection, configuration drift, and tenant isolation.", + ), + ModuleMaturityEvidence( + kind="test", + reference="tests/test_tabular_adapters.py", + summary="Exercises bounded CSV/XLSX parsing and the credential-governed read-only PostgreSQL adapter.", + ), ModuleMaturityEvidence( kind="test", reference="tests/test_sanctions_sources.py", @@ -120,7 +132,7 @@ ARCHITECTURE = ModuleArchitectureDeclaration( ), ), known_limits=( - "The executable generic datasource origin is an immutable tabular snapshot; database and arbitrary REST profiles remain future providers.", + "Tabular origins support immutable snapshots, exact managed CSV/XLSX versions, and read-only PostgreSQL tables; arbitrary REST and other database adapters remain future providers.", "Feed publication renders a governed document but does not yet push it to an external publishing endpoint.", "The generic governed runtime simulates deterministic mapping and validation; provider-specific live writes remain owned by explicit connector adapters.", ), @@ -158,7 +170,7 @@ EXTERNAL_PROVIDERS = ( ExternalProviderDeclaration( id=TABULAR_PROVIDER_ID, module_id=MODULE_ID, - label="Immutable tabular snapshot provider", + label="Governed tabular source providers", maturity="read", operations=("discover", "search", "read", "preview", "dry_run"), objects=( @@ -170,10 +182,10 @@ EXTERNAL_PROVIDERS = ( ), ), behavior=ProviderBehaviorDeclaration( - revision_tokens="Source fingerprints and immutable snapshot ids are retained.", - concurrency="Reads may require the expected fingerprint; snapshots never mutate in place.", - freshness="Snapshot acquisition time and source timestamp are exposed.", - health="Import validation and source-read failures are explicit.", + revision_tokens="Source fingerprints, exact file versions, SQL configuration revisions, credential revisions, and discovery revisions are retained.", + concurrency="Reads may require the expected fingerprint; file versions remain pinned and SQL configuration or schema drift blocks reads until reviewed refresh.", + freshness="Snapshot acquisition time, exact/current file versions, and live SQL discovery health are exposed.", + health="Import validation, Files access/integrity, SQL authentication, configuration drift, schema drift, timeouts, and source unavailability are explicit without exposing secrets.", max_read_items=1000, idempotency="Feed imports accept a caller request key and replay the same committed immutable source without refetching.", retry="Read-only acquisition may be retried only as a new deliberate request after a failed atomic operation.", @@ -187,7 +199,7 @@ EXTERNAL_PROVIDERS = ( classifications=("internal", "confidential", "restricted"), purposes=("governed import", "dataflow input", "evidence reconstruction"), retention="Datasources or the consuming domain supplies retention and hold policy.", - secret_handling="Generic snapshots contain no connector credential; transport credentials stay in credential envelopes.", + secret_handling="Snapshots and managed-file sources contain no connector credential; PostgreSQL uses a scoped Core credential-envelope reference and never stores or returns the resolved secret.", ), capability_names=( CAPABILITY_CONNECTORS_TABULAR_SOURCES, @@ -347,11 +359,13 @@ def _router(_context): def _provider(_context) -> SqlTabularSourceProvider: - return SqlTabularSourceProvider() + return SqlTabularSourceProvider(registry=getattr(_context, "registry", None)) def _datasource_origin_provider(_context) -> ConnectorDatasourceOriginProvider: - return ConnectorDatasourceOriginProvider() + return ConnectorDatasourceOriginProvider( + SqlTabularSourceProvider(registry=getattr(_context, "registry", None)) + ) def _sanctions_snapshot_provider( @@ -451,6 +465,14 @@ manifest = ModuleManifest( ), ModuleInterfaceProvider(name=CONNECTORS_DSAR_CAPABILITY, version="0.1.0"), ), + requires_interfaces=( + ModuleInterfaceRequirement( + name=CAPABILITY_FILES_TABULAR_CONTENT, + version_min="1.0.0", + version_max_exclusive="2.0.0", + optional=True, + ), + ), permissions=PERMISSIONS, role_templates=ROLE_TEMPLATES, route_factory=_router, @@ -646,11 +668,16 @@ manifest = ModuleManifest( "fingerprints, and bounded reads. Dataflow stores only opaque source " "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." + "projection, filter, aggregation, sorting, and pagination pushdown. " + "Immutable JSON/CSV snapshots remain available. Managed CSV/XLSX " + "sources use the optional Files capability, pin an exact authorized " + "version, apply archive and expansion limits, and require explicit " + "refresh before adopting a newer version. The PostgreSQL adapter uses " + "an active governed configuration and scoped Core credential envelope, " + "reflects simple schema/table identifiers, runs read-only bounded " + "projection and pagination, and blocks configuration, credential, or " + "schema drift until reviewed refresh. Credentials, endpoints, storage " + "keys, and raw file internals are never exposed through the origin." ), layer="available", documentation_types=("admin", "user"), diff --git a/src/govoplan_connectors/backend/provider_state.py b/src/govoplan_connectors/backend/provider_state.py index 3d11c23..05dabae 100644 --- a/src/govoplan_connectors/backend/provider_state.py +++ b/src/govoplan_connectors/backend/provider_state.py @@ -97,6 +97,13 @@ def _tabular_state( observed_at: datetime, ) -> ExternalProviderRuntimeState: active = source.status == "active" + live_origin = source.provider in {"managed_file", "postgresql"} + labels = { + "managed_file": "Exact managed-file origin", + "postgresql": "Live PostgreSQL origin", + "snapshot": "Immutable tabular snapshot", + } + label = labels.get(source.provider, "Tabular source") return ExternalProviderRuntimeState( provider_id=TABULAR_PROVIDER_ID, binding_ref=f"connectors:tabular-source:{source.id}", @@ -104,17 +111,20 @@ def _tabular_state( observed_at=observed_at, configured=True, active=active, - health="healthy" if active else "inactive", - freshness="not_applicable", + health=("unknown" if active and live_origin else "healthy" if active else "inactive"), + freshness="unknown" if active and live_origin else "not_applicable", conflict="not_applicable", recovery="ready" if active else "not_applicable", last_success_at=_aware(source.updated_at or source.created_at), detail=( - "Immutable tabular snapshot is available." + f"{label} is configured; live access and drift are checked on preview." + if active and live_origin + else f"{label} is available." if active - else "Immutable tabular snapshot is inactive." + else f"{label} is inactive." ), metrics={ + "provider": source.provider, "row_count": int(source.row_count), "byte_count": int(source.byte_count), "schema_version": int(source.schema_version), diff --git a/src/govoplan_connectors/backend/router.py b/src/govoplan_connectors/backend/router.py index 55fb6f8..4e3b373 100644 --- a/src/govoplan_connectors/backend/router.py +++ b/src/govoplan_connectors/backend/router.py @@ -32,6 +32,7 @@ from govoplan_connectors.backend.schemas import ( FeedImportRequest, FeedPublicationEntryPayload, FeedRenderPayload, + ManagedFileSourceCreateRequest, SanctionsAcquisitionRunListResponse, SanctionsAcquisitionRunResponse, SanctionsRefreshResponse, @@ -40,6 +41,7 @@ from govoplan_connectors.backend.schemas import ( SanctionsSourceListResponse, SanctionsSourceResponse, SnapshotCreateRequest, + SqlSourceCreateRequest, TabularColumnResponse, TabularHealthResponse, TabularPreviewDiagnosticResponse, @@ -460,12 +462,82 @@ def api_create_tabular_snapshot( return _source_response(source) +@router.post( + "/tabular-sources/files", + response_model=TabularSourceResponse, + status_code=status.HTTP_201_CREATED, +) +def api_create_managed_file_source( + payload: ManagedFileSourceCreateRequest, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> TabularSourceResponse: + _require_any_scope(principal, WRITE_SCOPE, ADMIN_SCOPE) + try: + source = provider.create_file_source( + session, + principal, + name=payload.name, + source_name=payload.source_name, + description=payload.description, + file_asset_id=payload.file_asset_id, + file_version_id=payload.file_version_id, + delimiter=payload.delimiter, + sheet_name=payload.sheet_name, + ) + except TabularSourceError as exc: + raise _http_error(exc) from exc + _audit_tabular_source_change( + session, + principal, + source, + operation="created", + ) + session.commit() + return _source_response(source) + + +@router.post( + "/tabular-sources/sql", + response_model=TabularSourceResponse, + status_code=status.HTTP_201_CREATED, +) +def api_create_sql_source( + payload: SqlSourceCreateRequest, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> TabularSourceResponse: + _require_any_scope(principal, WRITE_SCOPE, ADMIN_SCOPE) + try: + source = provider.create_sql_source( + session, + principal, + name=payload.name, + source_name=payload.source_name, + description=payload.description, + configuration_id=payload.configuration_id, + schema_name=payload.schema_name, + table_name=payload.table_name, + ) + except TabularSourceError as exc: + raise _http_error(exc) from exc + _audit_tabular_source_change( + session, + principal, + source, + operation="created", + ) + session.commit() + return _source_response(source) + + @router.get( "/tabular-sources/{source_id}/preview", response_model=TabularSourcePreviewResponse, ) def api_preview_tabular_source( source_id: str, + kind: Annotated[str, Query(pattern=r"^(snapshot|file|sql)$")] = "snapshot", 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), @@ -479,7 +551,7 @@ def api_preview_tabular_source( session, principal, request=TabularReadRequest( - source_ref=f"snapshot:{source_id}", + source_ref=f"{kind}:{source_id}", limit=limit, offset=offset, max_bytes=max_bytes, @@ -510,17 +582,47 @@ def api_preview_tabular_source( ) +@router.post( + "/tabular-sources/{source_id}/refresh", + response_model=TabularSourceResponse, +) +def api_refresh_tabular_source( + source_id: str, + kind: Annotated[str, Query(pattern=r"^(file|sql)$")], + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> TabularSourceResponse: + _require_any_scope(principal, WRITE_SCOPE, ADMIN_SCOPE) + try: + source = provider.refresh_source( + session, + principal, + source_ref=f"{kind}:{source_id}", + ) + except TabularSourceError as exc: + raise _http_error(exc) from exc + _audit_tabular_source_change( + session, + principal, + source, + operation="refreshed", + ) + session.commit() + return _source_response(source) + + @router.delete( "/tabular-sources/{source_id}", response_model=TabularSourceDeleteResponse, ) def api_delete_tabular_source( source_id: str, + kind: Annotated[str, Query(pattern=r"^(snapshot|file|sql)$")] = "snapshot", 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}" + source_ref = f"{kind}:{source_id}" try: source = provider.delete_snapshot( session, @@ -534,7 +636,7 @@ def api_delete_tabular_source( tenant_id=principal.tenant_id, user_id=getattr(principal.user, "id", None), api_key_id=principal.api_key_id, - action="connectors.tabular_snapshot.deleted", + action="connectors.tabular_source.deleted", object_type="connector_tabular_source", object_id=source_ref, details={"source_name": source.source_name, "fingerprint": source.fingerprint}, @@ -928,6 +1030,32 @@ def _source_response(source: TabularSource) -> TabularSourceResponse: ) +def _audit_tabular_source_change( + session: Session, + principal: ApiPrincipal, + source: TabularSource, + *, + operation: str, +) -> None: + audit_event( + session, + tenant_id=principal.tenant_id, + user_id=getattr(principal.user, "id", None), + api_key_id=principal.api_key_id, + action=f"connectors.tabular_source.{operation}", + object_type="connector_tabular_source", + object_id=source.ref, + details={ + "provider": source.provider, + "source_name": source.source_name, + "fingerprint": source.fingerprint, + "schema_version": source.schema_version, + "row_count": source.row_count, + "source_mode": source.source_mode, + }, + ) + + def _sanctions_snapshot_response( snapshot: SanctionsSnapshotReference, ) -> SanctionsSnapshotResponse: diff --git a/src/govoplan_connectors/backend/schemas.py b/src/govoplan_connectors/backend/schemas.py index 3c2d48e..f9e1b93 100644 --- a/src/govoplan_connectors/backend/schemas.py +++ b/src/govoplan_connectors/backend/schemas.py @@ -105,6 +105,40 @@ class SnapshotCreateRequest(BaseModel): return self +class ManagedFileSourceCreateRequest(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) + file_asset_id: str = Field(min_length=1, max_length=36) + file_version_id: str | None = Field(default=None, min_length=1, max_length=36) + delimiter: Literal[",", ";", "\t", "|"] = "," + sheet_name: str | None = Field(default=None, min_length=1, max_length=255) + + +class SqlSourceCreateRequest(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) + configuration_id: str = Field(min_length=1, max_length=36) + schema_name: str | None = Field( + default=None, + pattern=r"^[A-Za-z_][A-Za-z0-9_$-]{0,127}$", + ) + table_name: str = Field( + min_length=1, + max_length=128, + pattern=r"^[A-Za-z_][A-Za-z0-9_$-]{0,127}$", + ) + + class TabularColumnResponse(BaseModel): name: str data_type: str @@ -250,7 +284,9 @@ __all__ = [ "FeedEntryPayload", "FeedImportRequest", "FeedRenderPayload", + "ManagedFileSourceCreateRequest", "SnapshotCreateRequest", + "SqlSourceCreateRequest", "SanctionsAcquisitionRunListResponse", "SanctionsAcquisitionRunResponse", "SanctionsRefreshResponse", diff --git a/src/govoplan_connectors/backend/tabular_adapters.py b/src/govoplan_connectors/backend/tabular_adapters.py new file mode 100644 index 0000000..9cc7ded --- /dev/null +++ b/src/govoplan_connectors/backend/tabular_adapters.py @@ -0,0 +1,849 @@ +from __future__ import annotations + +import hashlib +import json +import math +import re +import zipfile +from collections.abc import Callable, Mapping, Sequence +from dataclasses import dataclass +from datetime import date, datetime +from decimal import Decimal +from io import BytesIO +from typing import Any + +from openpyxl import load_workbook +from sqlalchemy import ( + JSON, + BigInteger, + Boolean, + Date, + DateTime, + Float, + Integer, + LargeBinary, + MetaData, + Numeric, + String, + Table, + Text, + create_engine, + func, + select, +) +from sqlalchemy.engine import Engine, URL, make_url +from sqlalchemy.exc import SQLAlchemyError +from sqlalchemy.orm import Session + +from govoplan_core.auth import ApiPrincipal +from govoplan_core.core.files import ( + ManagedTabularFileAccessError, + ManagedTabularFileError, + ManagedTabularFileValidationError, + managed_tabular_file_provider, +) +from govoplan_core.core.tabular_sources import ( + TabularColumn, + TabularPreviewDiagnostic, + TabularPushdown, + TabularSourceHealth, + TabularSourceUnavailableError, + TabularSourceValidationError, + parse_tabular_csv, +) +from govoplan_core.security.credential_envelopes import ( + CredentialAccessContext, + CredentialEnvelopeError, + resolve_credential_envelope, +) +from govoplan_core.security.redaction import is_sensitive_key +from govoplan_connectors.backend.db.models import ConnectorConfiguration + + +MAX_FILE_BYTES = 5_000_000 +MAX_FILE_ROWS = 10_000 +MAX_FILE_COLUMNS = 500 +MAX_XLSX_ENTRIES = 5_000 +MAX_XLSX_EXPANDED_BYTES = 50_000_000 +MAX_XLSX_COMPRESSION_RATIO = 100 +POSTGRESQL_SCHEMES = frozenset({"postgresql", "postgresql+psycopg"}) +_IDENTIFIER = re.compile(r"^[A-Za-z_][A-Za-z0-9_$-]{0,127}$") + + +@dataclass(frozen=True, slots=True) +class TabularOriginInspection: + provider: str + schema: tuple[TabularColumn, ...] + fingerprint: str + row_count: int + byte_count: int + metadata: Mapping[str, object] + health: TabularSourceHealth + pushdown: TabularPushdown + rows: tuple[Mapping[str, object], ...] = () + diagnostics: tuple[TabularPreviewDiagnostic, ...] = () + + +@dataclass(frozen=True, slots=True) +class TabularOriginRead: + inspection: TabularOriginInspection + rows: tuple[Mapping[str, object], ...] + total_rows: int + diagnostics: tuple[TabularPreviewDiagnostic, ...] = () + + +class ManagedFileTabularAdapter: + def __init__(self, registry: object | None) -> None: + self._registry = registry + + def inspect( + self, + session: Session, + principal: ApiPrincipal, + *, + file_asset_id: str, + file_version_id: str | None, + delimiter: str = ",", + sheet_name: str | None = None, + ) -> TabularOriginInspection: + provider = managed_tabular_file_provider(self._registry) + if provider is None: + raise TabularSourceUnavailableError( + "Managed file sources require the Files module." + ) + try: + metadata = provider.get_tabular_file( + session, + principal, + file_asset_id=file_asset_id, + file_version_id=file_version_id, + ) + if metadata is None: + raise TabularSourceUnavailableError( + "Managed tabular file or version is unavailable." + ) + content = provider.read_tabular_file( + session, + principal, + file_asset_id=file_asset_id, + file_version_id=metadata.file_version_id, + max_bytes=MAX_FILE_BYTES, + ) + except ManagedTabularFileAccessError as exc: + raise TabularSourceUnavailableError( + "Managed tabular file access is no longer authorized." + ) from exc + except ManagedTabularFileValidationError as exc: + raise TabularSourceValidationError(str(exc)) from exc + except ManagedTabularFileError as exc: + raise TabularSourceUnavailableError(str(exc)) from exc + rows, resolved_sheet = parse_managed_tabular_content( + content.payload, + filename=content.file.filename, + content_type=content.file.content_type, + delimiter=delimiter, + sheet_name=sheet_name, + ) + schema = infer_tabular_schema(rows) + fingerprint = origin_fingerprint( + schema, + tokens=( + "managed_file", + content.file.file_asset_id, + content.file.file_version_id, + content.file.sha256, + resolved_sheet or "", + delimiter, + ), + ) + try: + current = provider.get_tabular_file( + session, + principal, + file_asset_id=file_asset_id, + ) + except ManagedTabularFileAccessError as exc: + raise TabularSourceUnavailableError( + "Managed tabular file access is no longer authorized." + ) from exc + except ManagedTabularFileError as exc: + raise TabularSourceUnavailableError(str(exc)) from exc + changed = bool( + current is not None + and current.file_version_id != content.file.file_version_id + ) + health = TabularSourceHealth( + status="warning" if changed else "healthy", + code=( + "files.newer_version_available" + if changed + else "files.exact_version_ready" + ), + summary=( + "A newer managed file version is available for explicit review." + if changed + else "The exact managed file version passed access and integrity checks." + ), + checked_at=content.file.updated_at, + details={ + "pinned_version_id": content.file.file_version_id, + "current_version_id": ( + current.file_version_id if current is not None else None + ), + }, + ) + return TabularOriginInspection( + provider="managed_file", + schema=schema, + fingerprint=fingerprint, + row_count=len(rows), + byte_count=len(content.payload), + metadata={ + "origin_kind": "managed_file", + "file_asset_id": content.file.file_asset_id, + "file_version_id": content.file.file_version_id, + "file_sha256": content.file.sha256, + "filename": content.file.filename, + "content_type": content.file.content_type, + "format": "xlsx" if _is_xlsx(content.file.filename, content.file.content_type) else "csv", + "delimiter": delimiter, + "sheet_name": resolved_sheet, + }, + health=health, + pushdown=TabularPushdown(projections=True, pagination=True), + rows=rows, + diagnostics=( + ( + TabularPreviewDiagnostic( + severity="warning", + code="files.newer_version_available", + message=( + "The preview remains pinned to the reviewed file version; " + "refresh the source to adopt the newer version." + ), + details=dict(health.details), + ), + ) + if changed + else () + ), + ) + + def read( + self, + session: Session, + principal: ApiPrincipal, + *, + metadata: Mapping[str, object], + ) -> TabularOriginRead: + inspection = self.inspect( + session, + principal, + file_asset_id=_required_metadata(metadata, "file_asset_id"), + file_version_id=_required_metadata(metadata, "file_version_id"), + delimiter=str(metadata.get("delimiter") or ","), + sheet_name=_optional_text(metadata.get("sheet_name")), + ) + expected_sha256 = _required_metadata(metadata, "file_sha256") + if inspection.metadata.get("file_sha256") != expected_sha256: + raise TabularSourceUnavailableError( + "Managed file content no longer matches its exact-version checksum." + ) + return TabularOriginRead( + inspection=inspection, + rows=inspection.rows, + total_rows=inspection.row_count, + diagnostics=inspection.diagnostics, + ) + + +class PostgresqlTabularAdapter: + """Read-only SQLAlchemy adapter with a production PostgreSQL allow-list.""" + + def __init__( + self, + *, + engine_factory: Callable[[URL], Engine] | None = None, + allow_sqlite_for_tests: bool = False, + ) -> None: + self._engine_factory = engine_factory or ( + lambda url: create_engine(url, pool_pre_ping=True) + ) + self._allow_sqlite_for_tests = allow_sqlite_for_tests + + def inspect( + self, + session: Session, + principal: ApiPrincipal, + *, + configuration_id: str, + table_name: str, + schema_name: str | None = None, + timeout_ms: int = 2_000, + ) -> TabularOriginInspection: + configuration, url, credential_revision = self._connection( + session, + principal, + configuration_id=configuration_id, + ) + table_name = _sql_identifier(table_name, "table") + schema_name = ( + _sql_identifier(schema_name, "schema") if schema_name else None + ) + engine = self._engine_factory(_bounded_connection_url(url, timeout_ms)) + try: + with engine.connect() as connection: + _configure_read_only(connection, url, timeout_ms) + table = Table( + table_name, + MetaData(), + schema=schema_name, + autoload_with=connection, + ) + schema = tuple(_sql_column(column) for column in table.columns) + if not schema: + raise TabularSourceValidationError( + "SQL tabular sources require at least one column." + ) + row_count = int( + connection.execute(select(func.count()).select_from(table)).scalar_one() + ) + except TabularSourceValidationError: + raise + except SQLAlchemyError as exc: + raise TabularSourceUnavailableError( + "SQL source discovery failed; verify the active configuration, credential, table, and provider health." + ) from exc + finally: + engine.dispose() + fingerprint = origin_fingerprint( + schema, + tokens=( + "postgresql", + configuration.id, + configuration.effective_hash, + credential_revision or "", + schema_name or "", + table_name, + ), + ) + return TabularOriginInspection( + provider="postgresql", + schema=schema, + fingerprint=fingerprint, + row_count=row_count, + byte_count=0, + metadata={ + "origin_kind": "sql", + "configuration_id": configuration.id, + "configuration_revision": configuration.resource_revision, + "configuration_hash": configuration.effective_hash, + "credential_revision": credential_revision, + "schema_name": schema_name, + "table_name": table_name, + }, + health=TabularSourceHealth( + status="healthy", + code="sql.source_ready", + summary="The governed PostgreSQL source is reachable and its schema was discovered.", + checked_at=configuration.updated_at, + details={ + "configuration_id": configuration.id, + "configuration_revision": configuration.resource_revision, + }, + ), + pushdown=TabularPushdown(projections=True, pagination=True), + ) + + def read( + self, + session: Session, + principal: ApiPrincipal, + *, + metadata: Mapping[str, object], + columns: Sequence[str], + offset: int, + limit: int, + timeout_ms: int, + ) -> TabularOriginRead: + configuration_id = _required_metadata(metadata, "configuration_id") + current_configuration, _current_url, current_credential_revision = ( + self._connection( + session, + principal, + configuration_id=configuration_id, + ) + ) + if current_configuration.effective_hash != metadata.get( + "configuration_hash" + ): + raise TabularSourceValidationError( + "The SQL connector configuration changed; refresh the source before previewing it." + ) + if current_credential_revision != metadata.get("credential_revision"): + raise TabularSourceValidationError( + "The SQL source credential changed; refresh the source before previewing it." + ) + inspection = self.inspect( + session, + principal, + configuration_id=configuration_id, + table_name=_required_metadata(metadata, "table_name"), + schema_name=_optional_text(metadata.get("schema_name")), + timeout_ms=timeout_ms, + ) + if inspection.fingerprint != metadata.get("discovery_fingerprint"): + raise TabularSourceValidationError( + "The SQL source schema drifted; refresh and review the source before previewing it." + ) + configuration, url, _credential_revision = self._connection( + session, + principal, + configuration_id=configuration_id, + ) + engine = self._engine_factory(_bounded_connection_url(url, timeout_ms)) + try: + with engine.connect() as connection: + _configure_read_only(connection, url, timeout_ms) + table = Table( + _required_metadata(metadata, "table_name"), + MetaData(), + schema=_optional_text(metadata.get("schema_name")), + autoload_with=connection, + ) + selected_names = tuple(dict.fromkeys(str(item) for item in columns)) + selected = ( + [table.c[name] for name in selected_names] + if selected_names + else list(table.columns) + ) + statement = select(*selected).offset(max(0, int(offset))).limit( + max(1, int(limit)) + ) + rows = tuple( + _json_row(dict(row._mapping)) + for row in connection.execute(statement) + ) + except KeyError as exc: + raise TabularSourceValidationError( + f"Unknown SQL source column: {exc.args[0]}" + ) from exc + except SQLAlchemyError as exc: + raise TabularSourceUnavailableError( + "SQL source preview failed or exceeded its provider budget." + ) from exc + finally: + engine.dispose() + return TabularOriginRead( + inspection=inspection, + rows=rows, + total_rows=inspection.row_count, + ) + + def _connection( + self, + session: Session, + principal: ApiPrincipal, + *, + configuration_id: str, + ) -> tuple[ConnectorConfiguration, URL, str | None]: + configuration = session.scalar( + select(ConnectorConfiguration).where( + ConnectorConfiguration.id == configuration_id, + ConnectorConfiguration.tenant_id == principal.tenant_id, + ) + ) + if configuration is None: + raise TabularSourceUnavailableError( + "SQL connector configuration is unavailable." + ) + if configuration.status != "active": + raise TabularSourceUnavailableError( + "SQL connector configuration is not active." + ) + endpoint = _optional_text(configuration.endpoint_url) + if not endpoint: + raise TabularSourceUnavailableError( + "SQL connector configuration has no endpoint." + ) + specification = dict(configuration.effective_configuration or {}) + provider = str(specification.get("provider") or "").strip().casefold() + protocol = str(specification.get("protocol") or "").strip().casefold() + if provider not in {"postgres", "postgresql", "sql"} or protocol not in { + "postgres", + "postgresql", + "sql", + }: + raise TabularSourceValidationError( + "The selected connector configuration is not a PostgreSQL tabular reader." + ) + try: + url = make_url(endpoint) + except (TypeError, ValueError) as exc: + raise TabularSourceValidationError( + "SQL connector endpoint is invalid." + ) from exc + if url.username or url.password: + raise TabularSourceValidationError( + "SQL connector endpoints must not contain credentials." + ) + if any(is_sensitive_key(key) for key in url.query): + raise TabularSourceValidationError( + "SQL connector endpoint query parameters must not contain credentials." + ) + allowed_schemes = set(POSTGRESQL_SCHEMES) + if self._allow_sqlite_for_tests: + allowed_schemes.update({"sqlite", "sqlite+pysqlite"}) + if url.drivername not in allowed_schemes: + raise TabularSourceValidationError( + "Only the governed PostgreSQL tabular adapter is enabled." + ) + credential_revision: str | None = None + credential_ref = _optional_text(configuration.credential_ref) + if credential_ref: + try: + credential = resolve_credential_envelope( + session, + credential_id=credential_ref, + context=CredentialAccessContext( + tenant_id=principal.tenant_id, + user_id=_principal_user_id(principal), + group_ids=frozenset(principal.principal.group_ids), + target_scope_type="tenant", + target_scope_id=principal.tenant_id, + module_id="connectors", + server_ref=endpoint, + ), + ) + except CredentialEnvelopeError as exc: + raise TabularSourceUnavailableError( + "SQL source credential is unavailable, inactive, or outside its allowed scope." + ) from exc + public_data = dict(credential.public_data) + secret_data = dict(credential.secret_data) + username = _optional_text( + public_data.get("username") + or secret_data.get("username") + or secret_data.get("user") + ) + password = _optional_text(secret_data.get("password")) + if url.drivername in POSTGRESQL_SCHEMES and (not username or not password): + raise TabularSourceUnavailableError( + "SQL source credential does not provide a username and password." + ) + url = url.set(username=username, password=password) + credential_revision = credential.revision + elif url.drivername in POSTGRESQL_SCHEMES: + raise TabularSourceUnavailableError( + "PostgreSQL tabular sources require a credential envelope reference." + ) + return configuration, url, credential_revision + + +def parse_managed_tabular_content( + payload: bytes, + *, + filename: str, + content_type: str | None, + delimiter: str, + sheet_name: str | None, +) -> tuple[tuple[Mapping[str, object], ...], str | None]: + if len(payload) > MAX_FILE_BYTES: + raise TabularSourceValidationError( + f"Managed tabular files are limited to {MAX_FILE_BYTES:,} bytes." + ) + if _is_xlsx(filename, content_type): + return _parse_xlsx(payload, sheet_name=sheet_name) + try: + text = payload.decode("utf-8-sig") + except UnicodeDecodeError as exc: + raise TabularSourceValidationError( + "Managed CSV files must use UTF-8 encoding." + ) from exc + return ( + tuple(parse_tabular_csv(text, delimiter=delimiter, max_rows=MAX_FILE_ROWS)), + None, + ) + + +def _parse_xlsx( + payload: bytes, + *, + sheet_name: str | None, +) -> tuple[tuple[Mapping[str, object], ...], str]: + _validate_xlsx_archive(payload) + try: + workbook = load_workbook( + BytesIO(payload), + read_only=True, + data_only=True, + keep_links=False, + ) + except Exception as exc: + raise TabularSourceValidationError( + "Managed XLSX content could not be parsed." + ) from exc + try: + available = tuple(workbook.sheetnames) + if not available: + raise TabularSourceValidationError( + "Managed XLSX content requires at least one worksheet." + ) + selected_name = _optional_text(sheet_name) or available[0] + if selected_name not in available: + raise TabularSourceValidationError( + f"Managed XLSX worksheet {selected_name!r} was not found." + ) + worksheet = workbook[selected_name] + iterator = worksheet.iter_rows(values_only=True) + try: + raw_headers = next(iterator) + except StopIteration as exc: + raise TabularSourceValidationError( + "Managed XLSX worksheet requires a header row." + ) from exc + headers = _xlsx_headers(raw_headers) + rows: list[Mapping[str, object]] = [] + for values in iterator: + if len(rows) >= MAX_FILE_ROWS: + raise TabularSourceValidationError( + f"Managed XLSX worksheets are limited to {MAX_FILE_ROWS:,} data rows." + ) + normalized = tuple(values[: len(headers)]) + if all(value in (None, "") for value in normalized): + continue + rows.append( + { + header: _json_value( + normalized[index] if index < len(normalized) else None + ) + for index, header in enumerate(headers) + } + ) + return tuple(rows), selected_name + finally: + workbook.close() + + +def _validate_xlsx_archive(payload: bytes) -> None: + try: + with zipfile.ZipFile(BytesIO(payload)) as archive: + entries = archive.infolist() + if len(entries) > MAX_XLSX_ENTRIES: + raise TabularSourceValidationError( + "Managed XLSX content contains too many archive entries." + ) + expanded = sum(max(0, item.file_size) for item in entries) + if expanded > MAX_XLSX_EXPANDED_BYTES: + raise TabularSourceValidationError( + "Managed XLSX content exceeds the expanded-size limit." + ) + compressed = sum(max(1, item.compress_size) for item in entries) + if expanded > compressed * MAX_XLSX_COMPRESSION_RATIO: + raise TabularSourceValidationError( + "Managed XLSX content exceeds the compression-ratio limit." + ) + if any( + item.filename.startswith(("/", "\\")) + or ".." in item.filename.replace("\\", "/").split("/") + for item in entries + ): + raise TabularSourceValidationError( + "Managed XLSX content contains an unsafe archive path." + ) + except zipfile.BadZipFile as exc: + raise TabularSourceValidationError( + "Managed XLSX content is not a valid workbook archive." + ) from exc + + +def _xlsx_headers(values: Sequence[object]) -> tuple[str, ...]: + if len(values) > MAX_FILE_COLUMNS: + raise TabularSourceValidationError( + f"Managed XLSX worksheets are limited to {MAX_FILE_COLUMNS:,} columns." + ) + headers = tuple(str(value or "").strip() for value in values) + while headers and not headers[-1]: + headers = headers[:-1] + if not headers or any(not header for header in headers): + raise TabularSourceValidationError( + "Managed XLSX content requires a non-empty header row." + ) + if len(set(headers)) != len(headers): + raise TabularSourceValidationError( + "Managed XLSX column names must be unique." + ) + return headers + + +def infer_tabular_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 origin_fingerprint( + schema: Sequence[TabularColumn], + *, + tokens: Sequence[str], +) -> str: + payload = { + "schema": [ + { + "name": column.name, + "data_type": column.data_type, + "nullable": column.nullable, + } + for column in schema + ], + "tokens": list(tokens), + } + return hashlib.sha256( + json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8") + ).hexdigest() + + +def _sql_column(column: Any) -> TabularColumn: + column_type = column.type + if isinstance(column_type, Boolean): + data_type = "boolean" + elif isinstance(column_type, (Integer, BigInteger)): + data_type = "integer" + elif isinstance(column_type, (Numeric, Float)): + data_type = "number" + elif isinstance(column_type, DateTime): + data_type = "datetime" + elif isinstance(column_type, Date): + data_type = "date" + elif isinstance(column_type, JSON): + data_type = "object" + elif isinstance(column_type, LargeBinary): + data_type = "binary" + elif isinstance(column_type, (String, Text)): + data_type = "string" + else: + data_type = str(column_type).casefold() + return TabularColumn( + name=str(column.name), + data_type=data_type, + nullable=bool(column.nullable), + ) + + +def _configure_read_only(connection: Any, url: URL, timeout_ms: int) -> None: + if url.drivername not in POSTGRESQL_SCHEMES: + return + effective_timeout = max(1, min(int(timeout_ms), 30_000)) + connection.exec_driver_sql("SET TRANSACTION READ ONLY") + connection.exec_driver_sql(f"SET LOCAL statement_timeout = {effective_timeout}") + + +def _bounded_connection_url(url: URL, timeout_ms: int) -> URL: + if url.drivername not in POSTGRESQL_SCHEMES: + return url + connect_timeout = max(1, math.ceil(min(int(timeout_ms), 30_000) / 1_000)) + return url.update_query_dict({"connect_timeout": str(connect_timeout)}) + + +def _sql_identifier(value: object, label: str) -> str: + normalized = str(value or "").strip() + if not _IDENTIFIER.fullmatch(normalized): + raise TabularSourceValidationError( + f"SQL {label} names must use a simple identifier." + ) + return normalized + + +def _required_metadata(metadata: Mapping[str, object], key: str) -> str: + value = _optional_text(metadata.get(key)) + if not value: + raise TabularSourceValidationError( + f"Tabular source metadata is missing {key}." + ) + return value + + +def _optional_text(value: object | None) -> str | None: + normalized = str(value or "").strip() + return normalized or None + + +def _principal_user_id(principal: ApiPrincipal) -> str | None: + return _optional_text(getattr(principal.user, "id", None)) + + +def _is_xlsx(filename: str, content_type: str | None) -> bool: + normalized_type = str(content_type or "").split(";", 1)[0].strip().casefold() + return str(filename or "").strip().casefold().endswith(".xlsx") or normalized_type == ( + "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" + ) + + +def _json_row(row: Mapping[str, object]) -> dict[str, object]: + return {str(key): _json_value(value) for key, value in row.items()} + + +def _json_value(value: object) -> object: + if value is None or isinstance(value, (str, bool, int)): + return value + if isinstance(value, float): + if not math.isfinite(value): + return str(value) + return value + if isinstance(value, Decimal): + return float(value) + if isinstance(value, (datetime, date)): + return value.isoformat() + if isinstance(value, bytes): + return value.hex() + if isinstance(value, (list, dict)): + return json.loads(json.dumps(value, default=str)) + return str(value) + + +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__.casefold() + + +__all__ = [ + "MAX_FILE_BYTES", + "MAX_FILE_ROWS", + "ManagedFileTabularAdapter", + "PostgresqlTabularAdapter", + "TabularOriginInspection", + "TabularOriginRead", + "infer_tabular_schema", + "origin_fingerprint", + "parse_managed_tabular_content", +] diff --git a/src/govoplan_connectors/backend/tabular_sources.py b/src/govoplan_connectors/backend/tabular_sources.py index 07b9bce..bc5f791 100644 --- a/src/govoplan_connectors/backend/tabular_sources.py +++ b/src/govoplan_connectors/backend/tabular_sources.py @@ -27,8 +27,14 @@ from govoplan_core.core.tabular_sources import ( TabularSourceValidationError, parse_tabular_csv, ) +from govoplan_core.core.runtime import get_registry from govoplan_core.db.base import utcnow from govoplan_connectors.backend.db.models import ConnectorTabularSource +from govoplan_connectors.backend.tabular_adapters import ( + ManagedFileTabularAdapter, + PostgresqlTabularAdapter, + TabularOriginInspection, +) READ_SCOPE = "connectors:source:read" @@ -42,8 +48,16 @@ MAX_READ_TIMEOUT_MS = 2_000 class SqlTabularSourceProvider: - def __init__(self, *, clock: Callable[[], float] = time.monotonic) -> None: + def __init__( + self, + *, + registry: object | None = None, + clock: Callable[[], float] = time.monotonic, + sql_adapter: PostgresqlTabularAdapter | None = None, + ) -> None: + self._registry = registry self._clock = clock + self._sql_adapter = sql_adapter or PostgresqlTabularAdapter() def list_sources( self, @@ -109,7 +123,6 @@ 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)) @@ -149,10 +162,51 @@ class SqlTabularSourceProvider: raise TabularSourceValidationError( f"Unknown source columns: {', '.join(unknown_columns)}" ) + started = self._clock() + source_rows: Sequence[Mapping[str, object]] + total_rows = item.row_count + source = _source_dto(item) + base_offset = offset + if item.provider == "managed_file": + read = self._file_adapter().read( + db, + api_principal, + metadata=item.metadata_, + ) + if read.inspection.fingerprint != item.fingerprint: + raise TabularSourceValidationError( + "The managed file source changed; refresh and review it before previewing." + ) + source_rows = read.rows[offset:] + total_rows = read.total_rows + source = _source_dto(item, inspection=read.inspection) + diagnostics.extend(read.diagnostics) + elif item.provider == "postgresql": + read = self._sql_adapter.read( + db, + api_principal, + metadata=item.metadata_, + columns=selected_columns, + offset=offset, + limit=limit + 1, + timeout_ms=timeout_ms, + ) + source_rows = read.rows + total_rows = read.total_rows + source = _source_dto(item, inspection=read.inspection) + diagnostics.extend(read.diagnostics) + else: + source_rows = item.rows[offset:] + + if int(max(0.0, self._clock() - started) * 1_000) >= timeout_ms: + raise TabularSourceUnavailableError( + "Tabular source preview exceeded its time budget." + ) + rows: list[dict[str, object]] = [] returned_bytes = 2 stopped_for = "" - for row in item.rows[offset:]: + for row in source_rows: if len(rows) >= limit: stopped_for = "rows" break @@ -203,10 +257,10 @@ class SqlTabularSourceProvider: ) ) return TabularReadResult( - source=_source_dto(item), + source=source, rows=tuple(rows), - total_rows=item.row_count, - truncated=offset + len(rows) < item.row_count, + total_rows=total_rows, + truncated=base_offset + len(rows) < total_rows, returned_bytes=returned_bytes, elapsed_ms=elapsed_ms, effective_row_limit=limit, @@ -215,6 +269,173 @@ class SqlTabularSourceProvider: diagnostics=tuple(diagnostics), ) + def create_file_source( + self, + session: object, + principal: object, + *, + name: str, + source_name: str, + file_asset_id: str, + file_version_id: str | None = None, + description: str | None = None, + delimiter: str = ",", + sheet_name: str | None = None, + ) -> TabularSource: + db, api_principal = _context(session, principal, WRITE_SCOPE) + inspection = self._file_adapter().inspect( + db, + api_principal, + file_asset_id=file_asset_id, + file_version_id=file_version_id, + delimiter=delimiter, + sheet_name=sheet_name, + ) + return self._create_origin( + db, + api_principal, + name=name, + source_name=source_name, + description=description, + inspection=inspection, + ) + + def create_sql_source( + self, + session: object, + principal: object, + *, + name: str, + source_name: str, + configuration_id: str, + table_name: str, + schema_name: str | None = None, + description: str | None = None, + ) -> TabularSource: + db, api_principal = _context(session, principal, WRITE_SCOPE) + inspection = self._sql_adapter.inspect( + db, + api_principal, + configuration_id=configuration_id, + table_name=table_name, + schema_name=schema_name, + timeout_ms=MAX_READ_TIMEOUT_MS, + ) + return self._create_origin( + db, + api_principal, + name=name, + source_name=source_name, + description=description, + inspection=inspection, + ) + + def refresh_source( + 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.") + if item.provider == "managed_file": + inspection = self._file_adapter().inspect( + db, + api_principal, + file_asset_id=str(item.metadata_.get("file_asset_id") or ""), + file_version_id=None, + delimiter=str(item.metadata_.get("delimiter") or ","), + sheet_name=_clean_optional(item.metadata_.get("sheet_name")), + ) + elif item.provider == "postgresql": + inspection = self._sql_adapter.inspect( + db, + api_principal, + configuration_id=str( + item.metadata_.get("configuration_id") or "" + ), + table_name=str(item.metadata_.get("table_name") or ""), + schema_name=_clean_optional(item.metadata_.get("schema_name")), + timeout_ms=MAX_READ_TIMEOUT_MS, + ) + else: + raise TabularSourceValidationError( + "Immutable snapshots cannot be refreshed; import a replacement snapshot." + ) + item.schema_version += 1 + item.schema_ = [_column_payload(column) for column in inspection.schema] + item.fingerprint = inspection.fingerprint + item.row_count = inspection.row_count + item.byte_count = inspection.byte_count + item.metadata_ = { + **dict(inspection.metadata), + "discovery_fingerprint": inspection.fingerprint, + } + item.updated_by = _actor_id(api_principal) + db.flush() + return _source_dto(item, inspection=inspection) + + def _create_origin( + self, + db: Session, + principal: ApiPrincipal, + *, + name: str, + source_name: str, + description: str | None, + inspection: TabularOriginInspection, + ) -> TabularSource: + clean_name = str(name or "").strip() + clean_source_name = str(source_name or "").strip() + if not clean_name or not clean_source_name: + raise TabularSourceValidationError( + "Tabular source name and source identifier are required." + ) + existing = db.scalar( + select(ConnectorTabularSource.id).where( + ConnectorTabularSource.tenant_id == principal.tenant_id, + ConnectorTabularSource.source_name == clean_source_name, + ) + ) + if existing is not None: + raise TabularSourceValidationError( + f"A tabular source named {clean_source_name!r} already exists." + ) + actor_id = _actor_id(principal) + item = ConnectorTabularSource( + tenant_id=principal.tenant_id, + provider=inspection.provider, + source_name=clean_source_name, + name=clean_name, + description=_clean_optional(description), + status="active", + schema_version=1, + schema_=[_column_payload(column) for column in inspection.schema], + rows=[], + fingerprint=inspection.fingerprint, + row_count=inspection.row_count, + byte_count=inspection.byte_count, + metadata_={ + **dict(inspection.metadata), + "discovery_fingerprint": inspection.fingerprint, + }, + created_by=actor_id, + updated_by=actor_id, + ) + db.add(item) + db.flush() + return _source_dto(item, inspection=inspection) + + def _file_adapter(self) -> ManagedFileTabularAdapter: + return ManagedFileTabularAdapter(self._registry or get_registry()) + def create_snapshot( self, session: object, @@ -347,21 +568,81 @@ def _source_record( tenant_id: str, source_ref: str, ) -> ConnectorTabularSource | None: - source_id = source_ref.removeprefix("snapshot:") - if not source_id or source_id == source_ref: + prefix, separator, source_id = str(source_ref or "").partition(":") + provider_by_prefix = { + "snapshot": "snapshot", + "file": "managed_file", + "sql": "postgresql", + } + provider = provider_by_prefix.get(prefix) + if not separator or not source_id or provider is None: return None return session.scalar( select(ConnectorTabularSource).where( ConnectorTabularSource.id == source_id, ConnectorTabularSource.tenant_id == tenant_id, + ConnectorTabularSource.provider == provider, ConnectorTabularSource.deleted_at.is_(None), ) ) -def _source_dto(item: ConnectorTabularSource) -> TabularSource: +def _source_dto( + item: ConnectorTabularSource, + *, + inspection: TabularOriginInspection | None = None, +) -> TabularSource: + prefix_by_provider = { + "snapshot": "snapshot", + "managed_file": "file", + "postgresql": "sql", + } + source_mode = { + "managed_file": "file_backed", + "postgresql": "live", + }.get(item.provider, "cached") + if inspection is not None: + pushdown = inspection.pushdown + health = inspection.health + elif item.provider == "managed_file": + pushdown = TabularPushdown(projections=True, pagination=True) + health = TabularSourceHealth( + status="unknown", + code="files.exact_version_not_checked", + summary="The exact managed file version will be re-authorized and integrity-checked on preview.", + checked_at=item.updated_at, + details={ + "file_version_id": item.metadata_.get("file_version_id"), + }, + ) + elif item.provider == "postgresql": + pushdown = TabularPushdown(projections=True, pagination=True) + health = TabularSourceHealth( + status="unknown", + code="sql.health_not_checked", + summary="The live SQL source will be checked against its pinned configuration, credential, and schema on preview.", + checked_at=item.updated_at, + details={ + "configuration_id": item.metadata_.get("configuration_id"), + "configuration_revision": item.metadata_.get( + "configuration_revision" + ), + }, + ) + else: + 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}, + ) return TabularSource( - ref=f"snapshot:{item.id}", + ref=f"{prefix_by_provider.get(item.provider, 'snapshot')}:{item.id}", provider=item.provider, source_name=item.source_name, name=item.name, @@ -374,18 +655,9 @@ 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}, - ), + source_mode=source_mode, + pushdown=pushdown, + health=health, ) diff --git a/tests/test_tabular_adapters.py b/tests/test_tabular_adapters.py new file mode 100644 index 0000000..968f36a --- /dev/null +++ b/tests/test_tabular_adapters.py @@ -0,0 +1,341 @@ +from __future__ import annotations + +import tempfile +import unittest +from io import BytesIO +from pathlib import Path +from types import SimpleNamespace + +from openpyxl import Workbook +from sqlalchemy import Column, Integer, MetaData, String, Table, create_engine +from sqlalchemy.orm import sessionmaker + +from govoplan_core.auth import ApiPrincipal +from govoplan_core.core.access import PrincipalRef +from govoplan_core.core.files import ( + CAPABILITY_FILES_TABULAR_CONTENT, + ManagedTabularFile, + ManagedTabularFileContent, +) +from govoplan_core.core.tabular_sources import ( + TabularSourceUnavailableError, + TabularSourceValidationError, +) +from govoplan_core.db.base import Base +from govoplan_connectors.backend.db.models import ( + ConnectorConfiguration, + ConnectorDefinition, +) +from govoplan_connectors.backend.tabular_adapters import ( + ManagedFileTabularAdapter, + PostgresqlTabularAdapter, + parse_managed_tabular_content, +) + + +def principal(tenant_id: str = "tenant-1") -> ApiPrincipal: + return ApiPrincipal( + principal=PrincipalRef( + account_id="account-1", + membership_id="membership-1", + tenant_id=tenant_id, + scopes=frozenset( + { + "connectors:source:read", + "connectors:source:write", + "files:file:read", + "files:file:download", + } + ), + ), + account=object(), + user=SimpleNamespace(id="user-1"), + ) + + +class _ManagedFiles: + def __init__(self, payload: bytes, *, filename: str = "cases.csv") -> None: + self.payload = payload + self.filename = filename + self.current_version_id = "version-2" + + def _file(self, version_id: str) -> ManagedTabularFile: + return ManagedTabularFile( + file_asset_id="asset-1", + file_version_id=version_id, + filename=self.filename, + display_path=f"Imports/{self.filename}", + content_type=( + "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" + if self.filename.endswith(".xlsx") + else "text/csv" + ), + size_bytes=len(self.payload), + sha256=("a" if version_id == "version-1" else "b") * 64, + current_version=version_id == self.current_version_id, + ) + + def list_tabular_files(self, session, principal, *, query="", limit=100): + del session, principal, query, limit + return (self._file(self.current_version_id),) + + def get_tabular_file( + self, + session, + principal, + *, + file_asset_id, + file_version_id=None, + ): + del session, principal + if file_asset_id != "asset-1": + return None + return self._file(file_version_id or self.current_version_id) + + def read_tabular_file( + self, + session, + principal, + *, + file_asset_id, + file_version_id, + max_bytes, + ): + del session, principal, file_asset_id + if len(self.payload) > max_bytes: + raise AssertionError("test payload exceeded adapter limit") + return ManagedTabularFileContent( + file=self._file(file_version_id), + payload=self.payload, + ) + + +class _Registry: + def __init__(self, provider) -> None: + self.provider = provider + + def has_capability(self, name): + return name == CAPABILITY_FILES_TABULAR_CONTENT + + def require_capability(self, name): + if not self.has_capability(name): + raise KeyError(name) + return self.provider + + +class ManagedFileTabularAdapterTests(unittest.TestCase): + def test_csv_is_exact_version_pinned_and_reports_newer_version(self) -> None: + adapter = ManagedFileTabularAdapter( + _Registry(_ManagedFiles(b"id,amount\n0012,12.5\n2,7\n")) + ) + + result = adapter.inspect( + object(), + principal(), + file_asset_id="asset-1", + file_version_id="version-1", + ) + + self.assertEqual("managed_file", result.provider) + self.assertEqual("version-1", result.metadata["file_version_id"]) + self.assertEqual("warning", result.health.status) + self.assertEqual("files.newer_version_available", result.health.code) + self.assertEqual("mixed", result.schema[0].data_type) + self.assertEqual(2, result.row_count) + self.assertTrue(result.pushdown.projections) + self.assertEqual("files.newer_version_available", result.diagnostics[0].code) + + def test_xlsx_uses_requested_sheet_and_closed_typed_schema(self) -> None: + workbook = Workbook() + first = workbook.active + first.title = "Ignore" + first.append(["ignored"]) + target = workbook.create_sheet("Monthly") + target.append(["case_id", "amount", "active"]) + target.append(["A-1", 12.5, True]) + target.append(["A-2", None, False]) + payload = BytesIO() + workbook.save(payload) + workbook.close() + + rows, sheet = parse_managed_tabular_content( + payload.getvalue(), + filename="monthly.xlsx", + content_type=( + "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" + ), + delimiter=",", + sheet_name="Monthly", + ) + + self.assertEqual("Monthly", sheet) + self.assertEqual("A-1", rows[0]["case_id"]) + self.assertIsNone(rows[1]["amount"]) + + def test_missing_files_capability_is_explicitly_unavailable(self) -> None: + with self.assertRaisesRegex( + TabularSourceUnavailableError, + "require the Files module", + ): + ManagedFileTabularAdapter(None).inspect( + object(), + principal(), + file_asset_id="asset-1", + file_version_id=None, + ) + + +class PostgresqlTabularAdapterTests(unittest.TestCase): + def setUp(self) -> None: + self.directory = tempfile.TemporaryDirectory( + prefix="govoplan-connectors-sql-adapter-" + ) + source_path = Path(self.directory.name) / "source.db" + self.source_url = f"sqlite+pysqlite:///{source_path}" + source_engine = create_engine(self.source_url) + metadata = MetaData() + self.table = Table( + "monthly_cases", + metadata, + Column("case_id", String, nullable=False), + Column("amount", Integer, nullable=True), + ) + metadata.create_all(source_engine) + with source_engine.begin() as connection: + connection.execute( + self.table.insert(), + ( + {"case_id": "A-1", "amount": 12}, + {"case_id": "A-2", "amount": None}, + ), + ) + source_engine.dispose() + + self.catalog_engine = create_engine("sqlite+pysqlite:///:memory:") + Base.metadata.create_all( + self.catalog_engine, + tables=( + ConnectorDefinition.__table__, + ConnectorConfiguration.__table__, + ), + ) + self.session = sessionmaker(bind=self.catalog_engine)() + definition = ConnectorDefinition( + id="definition-1", + tenant_id="tenant-1", + definition_key="postgresql.reader", + name="PostgreSQL reader", + status="active", + current_revision=1, + local_definition=True, + ) + self.configuration = ConnectorConfiguration( + id="configuration-1", + tenant_id="tenant-1", + definition_id=definition.id, + name="Monthly SQL", + status="active", + endpoint_url=self.source_url, + credential_ref=None, + base_definition_revision=1, + local_overrides={}, + protected_paths=[], + effective_configuration={"provider": "sql", "protocol": "sql"}, + effective_hash="configuration-hash-1", + resource_revision=1, + ambiguity_policy="manual_review", + ) + self.session.add_all((definition, self.configuration)) + self.session.commit() + self.adapter = PostgresqlTabularAdapter(allow_sqlite_for_tests=True) + + def tearDown(self) -> None: + self.session.close() + self.catalog_engine.dispose() + self.directory.cleanup() + + def test_discovers_and_reads_projection_from_governed_sql_configuration(self) -> None: + inspection = self.adapter.inspect( + self.session, + principal(), + configuration_id=self.configuration.id, + table_name="monthly_cases", + ) + metadata = { + **dict(inspection.metadata), + "discovery_fingerprint": inspection.fingerprint, + } + read = self.adapter.read( + self.session, + principal(), + metadata=metadata, + columns=("case_id",), + offset=1, + limit=10, + timeout_ms=2_000, + ) + + self.assertEqual("live", "live") + self.assertEqual(["case_id", "amount"], [item.name for item in inspection.schema]) + self.assertEqual(2, inspection.row_count) + self.assertEqual(({"case_id": "A-2"},), read.rows) + self.assertTrue(inspection.pushdown.projections) + self.assertFalse(inspection.pushdown.filters) + + def test_schema_drift_and_tenant_isolation_fail_closed(self) -> None: + inspection = self.adapter.inspect( + self.session, + principal(), + configuration_id=self.configuration.id, + table_name="monthly_cases", + ) + metadata = { + **dict(inspection.metadata), + "discovery_fingerprint": inspection.fingerprint, + } + engine = create_engine(self.source_url) + with engine.begin() as connection: + connection.exec_driver_sql( + "ALTER TABLE monthly_cases ADD COLUMN category TEXT" + ) + engine.dispose() + + with self.assertRaisesRegex(TabularSourceValidationError, "schema drifted"): + self.adapter.read( + self.session, + principal(), + metadata=metadata, + columns=(), + offset=0, + limit=10, + timeout_ms=2_000, + ) + with self.assertRaisesRegex( + TabularSourceUnavailableError, + "configuration is unavailable", + ): + self.adapter.inspect( + self.session, + principal("tenant-2"), + configuration_id=self.configuration.id, + table_name="monthly_cases", + ) + + def test_endpoint_query_credentials_are_rejected_before_connection(self) -> None: + self.configuration.endpoint_url = f"{self.source_url}?password=not-allowed" + self.session.commit() + + with self.assertRaisesRegex( + TabularSourceValidationError, + "query parameters must not contain credentials", + ): + self.adapter.inspect( + self.session, + principal(), + configuration_id=self.configuration.id, + table_name="monthly_cases", + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_tabular_origin_provider.py b/tests/test_tabular_origin_provider.py new file mode 100644 index 0000000..081b303 --- /dev/null +++ b/tests/test_tabular_origin_provider.py @@ -0,0 +1,313 @@ +from __future__ import annotations + +import tempfile +import unittest +from pathlib import Path +from types import SimpleNamespace + +from sqlalchemy import Column, Integer, MetaData, String, Table, create_engine +from sqlalchemy.orm import sessionmaker + +from govoplan_core.auth import ApiPrincipal +from govoplan_core.core.access import PrincipalRef +from govoplan_core.core.files import ( + CAPABILITY_FILES_TABULAR_CONTENT, + ManagedTabularFile, + ManagedTabularFileContent, +) +from govoplan_core.core.tabular_sources import ( + TabularReadRequest, + TabularSourceUnavailableError, + TabularSourceValidationError, +) +from govoplan_core.db.base import Base +from govoplan_core.security.credential_envelopes import CredentialEnvelope +from govoplan_connectors.backend.db.models import ( + ConnectorConfiguration, + ConnectorDefinition, + ConnectorTabularSource, +) +from govoplan_connectors.backend.tabular_adapters import PostgresqlTabularAdapter +from govoplan_connectors.backend.tabular_sources import SqlTabularSourceProvider + + +def principal(tenant_id: str = "tenant-1") -> ApiPrincipal: + return ApiPrincipal( + principal=PrincipalRef( + account_id="account-1", + membership_id="membership-1", + tenant_id=tenant_id, + scopes=frozenset( + { + "connectors:source:read", + "connectors:source:write", + "files:file:read", + "files:file:download", + } + ), + ), + account=object(), + user=SimpleNamespace(id="user-1"), + ) + + +class _ManagedFiles: + def __init__(self) -> None: + self.current = "version-1" + self.payloads = { + "version-1": b"id,name\n1,Ada\n2,Lin\n", + "version-2": b"id,name,active\n1,Ada,true\n2,Lin,false\n", + } + + def _metadata(self, version_id: str) -> ManagedTabularFile: + payload = self.payloads[version_id] + return ManagedTabularFile( + file_asset_id="asset-1", + file_version_id=version_id, + filename="people.csv", + display_path="Imports/people.csv", + content_type="text/csv", + size_bytes=len(payload), + sha256=("a" if version_id == "version-1" else "b") * 64, + current_version=version_id == self.current, + ) + + def list_tabular_files(self, session, principal, *, query="", limit=100): + del session, principal, query, limit + return (self._metadata(self.current),) + + def get_tabular_file( + self, + session, + principal, + *, + file_asset_id, + file_version_id=None, + ): + del session, principal + if file_asset_id != "asset-1": + return None + return self._metadata(file_version_id or self.current) + + def read_tabular_file( + self, + session, + principal, + *, + file_asset_id, + file_version_id, + max_bytes, + ): + del session, principal, file_asset_id, max_bytes + return ManagedTabularFileContent( + file=self._metadata(file_version_id), + payload=self.payloads[file_version_id], + ) + + +class _Registry: + def __init__(self, files) -> None: + self.files = files + + def has_capability(self, name): + return name == CAPABILITY_FILES_TABULAR_CONTENT + + def require_capability(self, name): + if not self.has_capability(name): + raise KeyError(name) + return self.files + + +class ConnectorTabularOriginProviderTests(unittest.TestCase): + def setUp(self) -> None: + self.catalog_engine = create_engine("sqlite+pysqlite:///:memory:") + Base.metadata.create_all( + self.catalog_engine, + tables=( + ConnectorTabularSource.__table__, + ConnectorDefinition.__table__, + ConnectorConfiguration.__table__, + CredentialEnvelope.__table__, + ), + ) + self.session = sessionmaker(bind=self.catalog_engine)() + self.files = _ManagedFiles() + self.directory = tempfile.TemporaryDirectory( + prefix="govoplan-connectors-origin-provider-" + ) + source_path = Path(self.directory.name) / "source.db" + self.sql_url = f"sqlite+pysqlite:///{source_path}" + source_engine = create_engine(self.sql_url) + metadata = MetaData() + source_table = Table( + "monthly_cases", + metadata, + Column("case_id", String, nullable=False), + Column("amount", Integer, nullable=True), + ) + metadata.create_all(source_engine) + with source_engine.begin() as connection: + connection.execute( + source_table.insert(), + ( + {"case_id": "A-1", "amount": 12}, + {"case_id": "A-2", "amount": None}, + ), + ) + source_engine.dispose() + definition = ConnectorDefinition( + id="definition-1", + tenant_id="tenant-1", + definition_key="postgresql.reader", + name="PostgreSQL reader", + status="active", + current_revision=1, + local_definition=True, + ) + self.configuration = ConnectorConfiguration( + id="configuration-1", + tenant_id="tenant-1", + definition_id=definition.id, + name="Monthly SQL", + status="active", + endpoint_url=self.sql_url, + credential_ref=None, + base_definition_revision=1, + local_overrides={}, + protected_paths=[], + effective_configuration={"provider": "sql", "protocol": "sql"}, + effective_hash="configuration-hash-1", + resource_revision=1, + ambiguity_policy="manual_review", + ) + self.session.add_all((definition, self.configuration)) + self.session.commit() + self.provider = SqlTabularSourceProvider( + registry=_Registry(self.files), + sql_adapter=PostgresqlTabularAdapter(allow_sqlite_for_tests=True), + ) + + def tearDown(self) -> None: + self.session.close() + self.catalog_engine.dispose() + self.directory.cleanup() + + def test_managed_file_source_stays_pinned_until_explicit_refresh(self) -> None: + created = self.provider.create_file_source( + self.session, + principal(), + name="Managed people", + source_name="managed_people", + file_asset_id="asset-1", + ) + self.session.commit() + self.files.current = "version-2" + + preview = self.provider.read_source( + self.session, + principal(), + request=TabularReadRequest(source_ref=created.ref, limit=10), + ) + refreshed = self.provider.refresh_source( + self.session, + principal(), + source_ref=created.ref, + ) + + self.assertTrue(created.ref.startswith("file:")) + self.assertEqual("file_backed", preview.source.source_mode) + self.assertEqual("version-1", preview.source.metadata["file_version_id"]) + self.assertEqual( + "files.newer_version_available", + preview.diagnostics[0].code, + ) + self.assertEqual("version-2", refreshed.metadata["file_version_id"]) + self.assertEqual("2", refreshed.schema_version) + self.assertEqual(3, len(refreshed.schema)) + + def test_sql_source_projects_and_blocks_changed_configuration_until_refresh(self) -> None: + created = self.provider.create_sql_source( + self.session, + principal(), + name="Monthly cases", + source_name="monthly_cases", + configuration_id=self.configuration.id, + table_name="monthly_cases", + ) + self.session.commit() + preview = self.provider.read_source( + self.session, + principal(), + request=TabularReadRequest( + source_ref=created.ref, + columns=("case_id",), + limit=1, + ), + ) + + self.assertTrue(created.ref.startswith("sql:")) + self.assertEqual("live", preview.source.source_mode) + self.assertEqual(({"case_id": "A-1"},), preview.rows) + self.assertEqual("preview.row_limit_reached", preview.diagnostics[-1].code) + self.assertIsNone( + self.provider.get_source( + self.session, + principal("tenant-2"), + source_ref=created.ref, + ) + ) + + self.configuration.effective_hash = "configuration-hash-2" + self.configuration.resource_revision = 2 + self.session.commit() + with self.assertRaisesRegex( + TabularSourceValidationError, + "configuration changed", + ): + self.provider.read_source( + self.session, + principal(), + request=TabularReadRequest(source_ref=created.ref), + ) + refreshed = self.provider.refresh_source( + self.session, + principal(), + source_ref=created.ref, + ) + self.assertEqual("2", refreshed.schema_version) + self.assertEqual("configuration-hash-2", refreshed.metadata["configuration_hash"]) + + def test_inactive_or_stale_sql_credentials_have_sanitized_diagnostics(self) -> None: + self.configuration.endpoint_url = ( + "postgresql+psycopg://db.example.invalid/govoplan" + ) + self.configuration.credential_ref = "missing-credential" + self.session.commit() + adapter = PostgresqlTabularAdapter() + + with self.assertRaisesRegex( + TabularSourceUnavailableError, + "credential is unavailable, inactive, or outside its allowed scope", + ): + adapter.inspect( + self.session, + principal(), + configuration_id=self.configuration.id, + table_name="monthly_cases", + ) + self.configuration.status = "disabled" + self.session.commit() + with self.assertRaisesRegex( + TabularSourceUnavailableError, + "configuration is not active", + ): + adapter.inspect( + self.session, + principal(), + configuration_id=self.configuration.id, + table_name="monthly_cases", + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/webui/package.json b/webui/package.json index 613c5f7..d300c19 100644 --- a/webui/package.json +++ b/webui/package.json @@ -1,6 +1,6 @@ { "name": "@govoplan/connectors-webui", - "version": "0.1.19", + "version": "0.1.20", "private": true, "type": "module", "main": "src/index.ts",