commit 1cb6228442af2191e1edafdf3fedc7e1e9ba2b65 Author: Albrecht Degering Date: Tue Jul 28 12:43:02 2026 +0200 feat: initialize governed datasources module diff --git a/.gitea/ISSUE_TEMPLATE/bug_report.md b/.gitea/ISSUE_TEMPLATE/bug_report.md new file mode 100644 index 0000000..038d41b --- /dev/null +++ b/.gitea/ISSUE_TEMPLATE/bug_report.md @@ -0,0 +1,35 @@ +--- +name: "Bug" +about: "Report a reproducible defect, regression, or incorrect behavior" +title: "[Bug] " +labels: + - type/bug + - status/triage + - module/datasources +--- + +## Scope + +- Repository: +- Area/module: +- Affected version or commit: + +## Behavior + +Expected: + +Actual: + +## Reproduction + +1. +2. +3. + +## Evidence + +Logs, screenshots, traces, or failing test output: + +## Verification Target + +Command or workflow that should pass when fixed: diff --git a/.gitea/ISSUE_TEMPLATE/config.yaml b/.gitea/ISSUE_TEMPLATE/config.yaml new file mode 100644 index 0000000..3ba13e0 --- /dev/null +++ b/.gitea/ISSUE_TEMPLATE/config.yaml @@ -0,0 +1 @@ +blank_issues_enabled: false diff --git a/.gitea/ISSUE_TEMPLATE/docs_workflow.md b/.gitea/ISSUE_TEMPLATE/docs_workflow.md new file mode 100644 index 0000000..e46c70e --- /dev/null +++ b/.gitea/ISSUE_TEMPLATE/docs_workflow.md @@ -0,0 +1,27 @@ +--- +name: "Docs / workflow" +about: "Request documentation, process, or developer workflow changes" +title: "[Docs] " +labels: + - type/docs + - status/triage + - module/datasources + - area/docs +--- + +## Scope + +- Repository: +- Document or workflow: + +## Current State + +What is missing, unclear, duplicated, or stale? + +## Desired State + +What should the docs or workflow make clear? + +## Verification Target + +How should this be checked? diff --git a/.gitea/ISSUE_TEMPLATE/feature_request.md b/.gitea/ISSUE_TEMPLATE/feature_request.md new file mode 100644 index 0000000..130b5cb --- /dev/null +++ b/.gitea/ISSUE_TEMPLATE/feature_request.md @@ -0,0 +1,32 @@ +--- +name: "Feature" +about: "Propose new user-visible behavior or platform capability" +title: "[Feature] " +labels: + - type/feature + - status/triage + - module/datasources +--- + +## Problem + +What user, operator, or developer problem should this solve? + +## Proposed Capability + +What should exist when this is done? + +## Ownership + +- Owning repository: +- Related module repositories: +- Extension point or integration boundary: + +## Acceptance Criteria + +- [ ] +- [ ] + +## Verification Target + +Command, scenario, or UI flow that should prove completion: diff --git a/.gitea/ISSUE_TEMPLATE/task.md b/.gitea/ISSUE_TEMPLATE/task.md new file mode 100644 index 0000000..776b04c --- /dev/null +++ b/.gitea/ISSUE_TEMPLATE/task.md @@ -0,0 +1,28 @@ +--- +name: "Task" +about: "Track implementation, maintenance, or migration work" +title: "[Task] " +labels: + - type/task + - status/triage + - module/datasources +--- + +## Objective + +What needs to be completed? + +## Scope + +- Owning repository: +- In-scope: +- Out-of-scope: + +## Checklist + +- [ ] +- [ ] + +## Verification Target + +Command or manual check: diff --git a/.gitea/ISSUE_TEMPLATE/tech_debt.md b/.gitea/ISSUE_TEMPLATE/tech_debt.md new file mode 100644 index 0000000..c7cab52 --- /dev/null +++ b/.gitea/ISSUE_TEMPLATE/tech_debt.md @@ -0,0 +1,25 @@ +--- +name: "Tech debt" +about: "Track cleanup, refactoring, risk reduction, or deferred engineering work" +title: "[Debt] " +labels: + - type/debt + - status/triage + - module/datasources +--- + +## Current Cost + +What does this make harder, riskier, slower, or more fragile? + +## Desired Shape + +What should the code, tests, or architecture look like afterwards? + +## Constraints + +What behavior, compatibility, or module boundary must be preserved? + +## Verification Target + +Focused checks that should pass: diff --git a/.gitea/PULL_REQUEST_TEMPLATE.md b/.gitea/PULL_REQUEST_TEMPLATE.md new file mode 100644 index 0000000..1984736 --- /dev/null +++ b/.gitea/PULL_REQUEST_TEMPLATE.md @@ -0,0 +1,15 @@ +## Issue + +Closes # + +## Summary + +- + +## Verification + +- + +## Notes + +Follow-up issues: diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..a1c41d4 --- /dev/null +++ b/.gitignore @@ -0,0 +1,168 @@ +# Node +logs +*.log +npm-debug.log* +yarn-debug.log* +yarn-error.log* +.pnpm-debug.log* +report.[0-9]*.[0-9]*.[0-9]*.[0-9]*.json +pids +*.pid +*.seed +*.pid.lock +lib-cov +coverage +*.lcov +.nyc_output +.grunt +bower_components +.lock-wscript +build/Release +node_modules/ +jspm_packages/ +web_modules/ +*.tsbuildinfo +.npm +.eslintcache +.stylelintcache +.rpt2_cache/ +.rts2_cache_cjs/ +.rts2_cache_es/ +.rts2_cache_umd/ +.node_repl_history +*.tgz +.yarn-integrity +.env +.env.development.local +.env.test.local +.env.production.local +.env.local +.cache +.parcel-cache +.next +out +.nuxt +dist +.vuepress/dist +.temp +**/.vitepress/dist +**/.vitepress/cache +.docusaurus +.serverless/ +.fusebox/ +.dynamodb/ +.tern-port +.vscode-test +.yarn/cache +.yarn/unplugged +.yarn/build-state.yml +.yarn/install-state.gz +.pnp.* +.component-test-build/ +.file-drop-test-build/ +.module-test-build/ +.policy-test-build/ +.template-preview-test-build/ +.import-test-build/ +webui/.component-test-build/ +webui/.file-drop-test-build/ +webui/.module-test-build/ +webui/.policy-test-build/ +webui/.template-preview-test-build/ +webui/.import-test-build/ +audit-reports/ + +# Python +__pycache__/ +*.py[cod] +*$py.class +*.so +.Python +build/ +develop-eggs/ +dist/ +downloads/ +eggs/ +.eggs/ +lib/ +lib64/ +parts/ +sdist/ +var/ +wheels/ +share/python-wheels/ +*.egg-info/ +.installed.cfg +*.egg +MANIFEST +*.manifest +*.spec +pip-log.txt +pip-delete-this-directory.txt +htmlcov/ +.tox/ +.nox/ +.coverage +.coverage.* +.cache +nosetests.xml +coverage.xml +*.cover +*.py,cover +.hypothesis/ +.pytest_cache/ +cover/ +*.mo +*.pot +local_settings.py +db.sqlite3 +db.sqlite3-journal +instance/ +.webassets-cache +.scrapy +docs/_build/ +.pybuilder/ +target/ +.ipynb_checkpoints +profile_default/ +ipython_config.py +.pdm.toml +.pdm-python +.pdm-build/ +__pypackages__/ +celerybeat-schedule +celerybeat.pid +*.sage.py +.env +.venv +env/ +venv/ +ENV/ +env.bak/ +venv.bak/ +.spyderproject +.spyproject +.ropeproject +/site +.mypy_cache/ +.dmypy.json +dmypy.json +.pyre/ +.pytype/ +cython_debug/ +.ruff_cache/ +.pypirc + +# Editors +.vscode/* +!.vscode/settings.json +!.vscode/tasks.json +!.vscode/launch.json +!.vscode/extensions.json +!.vscode/*.code-snippets +.history/ +*.vsix + +# GovOPlaN local state +*.db +runtime/ diff --git a/README.md b/README.md new file mode 100644 index 0000000..3f3b09a --- /dev/null +++ b/README.md @@ -0,0 +1,25 @@ +# govoplan-datasources + + +**Repository type:** module (platform). + + +`govoplan-datasources` is GovOPlaN's provider-neutral data access layer. It owns +the tenant catalogue, lifecycle, staging, immutable materializations, frozen +states, schema and fingerprint metadata, provenance, and bounded reads. + +The module deliberately separates two layers: + +- **Datasources** describe governed data that modules can consume consistently. +- **Connectors** acquire data from external systems and expose provider origins. + +Dataflow, Workflow, Reporting, Risk Compliance, and other consumers depend on +Datasource contracts rather than connector implementations. + +The first executable slice supports tabular static uploads, connector-backed +live and cached sources, staging and promotion, refresh, immutable snapshots, +explicit frozen states, previews, and retirement. The contracts already model +database, HTTP/REST, directory, file, feed, document, binary, directory, and +stream sources so providers can be added without changing consumers. + +See [docs/CONCEPT.md](docs/CONCEPT.md) for ownership and lifecycle details. diff --git a/docs/CONCEPT.md b/docs/CONCEPT.md new file mode 100644 index 0000000..99421a8 --- /dev/null +++ b/docs/CONCEPT.md @@ -0,0 +1,63 @@ +# GovOPlaN Datasources + +## Boundary + +Datasources owns the governed identity and lifecycle of consumable data. +Connectors owns external protocols, connection profiles, credentials, provider +health, discovery, and source-side query pushdown. + +An external system is therefore represented in two steps: + +1. a connector exposes a provider-specific **origin**; +2. Datasources registers that origin as a governed live or cached datasource. + +Static uploads are staged and promoted directly by Datasources. Other modules +only retain opaque datasource and materialization references. + +## Modes + +- `live`: reads the external origin at request time. A live source can still be + frozen to preserve evidence for a run. +- `cached`: refreshes the origin into immutable local materializations. Reads + use the current materialization unless a specific frozen revision is chosen. +- `static`: promotes supplied rows into immutable materializations and never + refreshes an external origin. + +Materializations are append-only. Changing source data creates a new revision; +old revisions remain addressable for reproducibility. + +## Staging + +Staging is the inspection boundary before data becomes generally consumable. +A stage records schema, row and byte counts, fingerprint, validation result, +provenance, and intended target. Promotion either creates a datasource or adds +a new immutable materialization to an existing compatible datasource. + +The first slice stores bounded tabular JSON/CSV stages. Future providers may +stage file references, object-store blobs, directory snapshots, or streaming +checkpoints through the same lifecycle contract. + +## Consumer Contract + +Consumers request: + +- the current state; +- a direct live read where policy permits; +- a particular materialization; or +- the latest frozen state. + +Reads are bounded and tenant-scoped. Schema and expected fingerprints allow a +Dataflow or Workflow definition to detect changed inputs before execution. + +## Next Providers + +Connector providers should cover: + +- SQL databases with governed query and pushdown policies; +- REST and generic HTTP requests; +- LDAP and directory services; +- managed files and watched directories; +- feeds such as RSS/Atom; +- object storage and public-sector application APIs. + +Credential and connection policy remain outside this module. diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..2ed5a56 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,22 @@ +[build-system] +requires = ["setuptools>=69", "wheel"] +build-backend = "setuptools.build_meta" + +[project] +name = "govoplan-datasources" +version = "0.1.14" +description = "Governed datasource catalogue, staging, and materialization lifecycle for GovOPlaN." +readme = "README.md" +requires-python = ">=3.12" +license = "AGPL-3.0-or-later" +authors = [{ name = "GovOPlaN" }] +dependencies = ["govoplan-core>=0.1.14"] + +[tool.setuptools.packages.find] +where = ["src"] + +[tool.setuptools.package-data] +govoplan_datasources = ["py.typed"] + +[project.entry-points."govoplan.modules"] +datasources = "govoplan_datasources.backend.manifest:get_manifest" diff --git a/src/govoplan_datasources/__init__.py b/src/govoplan_datasources/__init__.py new file mode 100644 index 0000000..268031c --- /dev/null +++ b/src/govoplan_datasources/__init__.py @@ -0,0 +1,3 @@ +"""GovOPlaN Datasources module.""" + +__version__ = "0.1.14" diff --git a/src/govoplan_datasources/backend/__init__.py b/src/govoplan_datasources/backend/__init__.py new file mode 100644 index 0000000..71162ef --- /dev/null +++ b/src/govoplan_datasources/backend/__init__.py @@ -0,0 +1 @@ +"""Datasources backend.""" diff --git a/src/govoplan_datasources/backend/db/__init__.py b/src/govoplan_datasources/backend/db/__init__.py new file mode 100644 index 0000000..6d68dab --- /dev/null +++ b/src/govoplan_datasources/backend/db/__init__.py @@ -0,0 +1,13 @@ +"""Datasource database models.""" + +from govoplan_datasources.backend.db.models import ( + DatasourceMaterializationRecord, + DatasourceRecord, + DatasourceStageRecord, +) + +__all__ = [ + "DatasourceMaterializationRecord", + "DatasourceRecord", + "DatasourceStageRecord", +] diff --git a/src/govoplan_datasources/backend/db/models.py b/src/govoplan_datasources/backend/db/models.py new file mode 100644 index 0000000..bf74d79 --- /dev/null +++ b/src/govoplan_datasources/backend/db/models.py @@ -0,0 +1,217 @@ +from __future__ import annotations + +import uuid +from datetime import datetime +from typing import Any + +from sqlalchemy import ( + DateTime, + ForeignKey, + Index, + Integer, + JSON, + String, + Text, + UniqueConstraint, +) +from sqlalchemy.orm import Mapped, mapped_column, relationship + +from govoplan_core.db.base import Base, TimestampMixin + + +def new_uuid() -> str: + return str(uuid.uuid4()) + + +class DatasourceRecord(Base, TimestampMixin): + __tablename__ = "datasource_catalogue" + __table_args__ = ( + UniqueConstraint("tenant_id", "source_name", name="uq_datasource_source_name"), + Index("ix_datasource_catalogue_tenant_status", "tenant_id", "status"), + Index("ix_datasource_catalogue_tenant_mode", "tenant_id", "mode"), + Index("ix_datasource_catalogue_origin", "tenant_id", "provider", "provider_ref"), + ) + + id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid) + tenant_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True) + source_name: Mapped[str] = mapped_column(String(120), nullable=False) + name: Mapped[str] = mapped_column(String(300), nullable=False) + description: Mapped[str | None] = mapped_column(Text) + kind: Mapped[str] = mapped_column(String(30), nullable=False, index=True) + mode: Mapped[str] = mapped_column(String(20), nullable=False, index=True) + shape: Mapped[str] = mapped_column(String(30), nullable=False, index=True) + status: Mapped[str] = mapped_column(String(30), default="active", nullable=False, index=True) + provider: Mapped[str | None] = mapped_column(String(100), nullable=True, index=True) + provider_ref: Mapped[str | None] = mapped_column(String(500), nullable=True) + schema_version: Mapped[int] = mapped_column(Integer, default=1, nullable=False) + schema_: Mapped[list[dict[str, Any]]] = mapped_column( + "schema", + JSON, + default=list, + nullable=False, + ) + fingerprint: Mapped[str] = mapped_column(String(64), default="", nullable=False, index=True) + current_materialization_id: Mapped[str | None] = mapped_column( + String(36), + nullable=True, + index=True, + ) + row_count: Mapped[int | None] = mapped_column(Integer, nullable=True) + byte_count: Mapped[int | None] = mapped_column(Integer, nullable=True) + provenance_: Mapped[dict[str, Any]] = mapped_column( + "provenance", + JSON, + default=dict, + nullable=False, + ) + metadata_: Mapped[dict[str, Any]] = mapped_column( + "metadata", + JSON, + default=dict, + nullable=False, + ) + created_by: Mapped[str | None] = mapped_column(String(255), nullable=True, index=True) + updated_by: Mapped[str | None] = mapped_column(String(255), nullable=True, index=True) + deleted_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), + nullable=True, + index=True, + ) + + materializations: Mapped[list["DatasourceMaterializationRecord"]] = relationship( + back_populates="datasource", + cascade="all, delete-orphan", + order_by="DatasourceMaterializationRecord.revision", + ) + + +class DatasourceMaterializationRecord(Base, TimestampMixin): + __tablename__ = "datasource_materializations" + __table_args__ = ( + UniqueConstraint( + "datasource_id", + "revision", + name="uq_datasource_materialization_revision", + ), + Index( + "ix_datasource_materializations_tenant_source", + "tenant_id", + "datasource_id", + ), + Index( + "ix_datasource_materializations_frozen", + "tenant_id", + "datasource_id", + "frozen_at", + ), + ) + + id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid) + tenant_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True) + datasource_id: Mapped[str] = mapped_column( + ForeignKey("datasource_catalogue.id", ondelete="CASCADE"), + nullable=False, + index=True, + ) + revision: Mapped[int] = mapped_column(Integer, nullable=False) + state: Mapped[str] = mapped_column(String(30), default="published", nullable=False, index=True) + schema_version: Mapped[int] = mapped_column(Integer, default=1, nullable=False) + schema_: Mapped[list[dict[str, Any]]] = mapped_column( + "schema", + JSON, + default=list, + nullable=False, + ) + rows: Mapped[list[dict[str, Any]]] = mapped_column(JSON, default=list, nullable=False) + fingerprint: Mapped[str] = mapped_column(String(64), nullable=False, index=True) + row_count: Mapped[int] = mapped_column(Integer, nullable=False) + byte_count: Mapped[int] = mapped_column(Integer, nullable=False) + frozen_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + frozen_label: Mapped[str | None] = mapped_column(String(300), nullable=True) + source_timestamp: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), + nullable=True, + ) + provenance_: Mapped[dict[str, Any]] = mapped_column( + "provenance", + JSON, + default=dict, + nullable=False, + ) + metadata_: Mapped[dict[str, Any]] = mapped_column( + "metadata", + JSON, + default=dict, + nullable=False, + ) + created_by: Mapped[str | None] = mapped_column(String(255), nullable=True, index=True) + + datasource: Mapped[DatasourceRecord] = relationship(back_populates="materializations") + + +class DatasourceStageRecord(Base, TimestampMixin): + __tablename__ = "datasource_stages" + __table_args__ = ( + Index("ix_datasource_stages_tenant_state", "tenant_id", "state"), + Index("ix_datasource_stages_target", "tenant_id", "target_datasource_id"), + ) + + id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid) + tenant_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True) + target_datasource_id: Mapped[str | None] = mapped_column( + ForeignKey("datasource_catalogue.id", ondelete="SET NULL"), + nullable=True, + index=True, + ) + name: Mapped[str] = mapped_column(String(300), nullable=False) + source_name: Mapped[str] = mapped_column(String(120), nullable=False) + description: Mapped[str | None] = mapped_column(Text) + kind: Mapped[str] = mapped_column(String(30), nullable=False) + mode: Mapped[str] = mapped_column(String(20), nullable=False) + shape: Mapped[str] = mapped_column(String(30), nullable=False) + state: Mapped[str] = mapped_column(String(30), default="ready", nullable=False, index=True) + provider: Mapped[str | None] = mapped_column(String(100), nullable=True) + provider_ref: Mapped[str | None] = mapped_column(String(500), nullable=True) + schema_: Mapped[list[dict[str, Any]]] = mapped_column( + "schema", + JSON, + default=list, + nullable=False, + ) + rows: Mapped[list[dict[str, Any]]] = mapped_column(JSON, default=list, nullable=False) + fingerprint: Mapped[str] = mapped_column(String(64), nullable=False, index=True) + row_count: Mapped[int] = mapped_column(Integer, nullable=False) + byte_count: Mapped[int] = mapped_column(Integer, nullable=False) + validation_: Mapped[dict[str, Any]] = mapped_column( + "validation", + JSON, + default=dict, + nullable=False, + ) + provenance_: Mapped[dict[str, Any]] = mapped_column( + "provenance", + JSON, + default=dict, + nullable=False, + ) + metadata_: Mapped[dict[str, Any]] = mapped_column( + "metadata", + JSON, + default=dict, + nullable=False, + ) + promoted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + promoted_materialization_id: Mapped[str | None] = mapped_column( + String(36), + nullable=True, + index=True, + ) + created_by: Mapped[str | None] = mapped_column(String(255), nullable=True, index=True) + + +__all__ = [ + "DatasourceMaterializationRecord", + "DatasourceRecord", + "DatasourceStageRecord", + "new_uuid", +] diff --git a/src/govoplan_datasources/backend/manifest.py b/src/govoplan_datasources/backend/manifest.py new file mode 100644 index 0000000..76c75b2 --- /dev/null +++ b/src/govoplan_datasources/backend/manifest.py @@ -0,0 +1,287 @@ +from __future__ import annotations + +from pathlib import Path + +from govoplan_core.core.access import ( + CAPABILITY_AUTH_PERMISSION_EVALUATOR, + CAPABILITY_AUTH_PRINCIPAL_RESOLVER, +) +from govoplan_core.core.datasources import ( + CAPABILITY_DATASOURCE_CATALOGUE, + CAPABILITY_DATASOURCE_LIFECYCLE, + CAPABILITY_DATASOURCE_ORIGINS, +) +from govoplan_core.core.module_guards import ( + drop_table_retirement_provider, + persistent_table_uninstall_guard, +) +from govoplan_core.core.modules import ( + DocumentationTopic, + FrontendModule, + MigrationSpec, + ModuleContext, + ModuleInterfaceProvider, + ModuleInterfaceRequirement, + ModuleManifest, + NavItem, + PermissionDefinition, + RoleTemplate, +) +from govoplan_core.db.base import Base +from govoplan_datasources.backend.db import models as datasource_models +from govoplan_datasources.backend.service import ( + ADMIN_SCOPE, + CATALOGUE_READ_SCOPE, + SOURCE_WRITE_SCOPE, + STAGE_WRITE_SCOPE, + SqlDatasourceProvider, +) + + +MODULE_ID = "datasources" +MODULE_NAME = "Datasources" +MODULE_VERSION = "0.1.14" +DATASOURCE_INTERFACE_VERSION = "0.1.0" + + +def _permission(scope: str, label: str, description: str) -> PermissionDefinition: + module_id, resource, action = scope.split(":", 2) + return PermissionDefinition( + scope=scope, + label=label, + description=description, + category="Datasources", + level="tenant", + module_id=module_id, + resource=resource, + action=action, + ) + + +PERMISSIONS = ( + _permission( + CATALOGUE_READ_SCOPE, + "View datasources", + "Discover and preview policy-visible datasources and materializations.", + ), + _permission( + SOURCE_WRITE_SCOPE, + "Manage datasources", + "Register, refresh, freeze, and retire governed datasources.", + ), + _permission( + STAGE_WRITE_SCOPE, + "Stage datasource content", + "Upload, validate, inspect, and promote bounded datasource stages.", + ), + _permission( + ADMIN_SCOPE, + "Administer datasources", + "Manage every tenant datasource, stage, materialization, and lifecycle policy.", + ), +) + +ROLE_TEMPLATES = ( + RoleTemplate( + slug="datasource_manager", + name="Datasource manager", + description="Register, stage, refresh, freeze, and retire governed data.", + permissions=( + CATALOGUE_READ_SCOPE, + SOURCE_WRITE_SCOPE, + STAGE_WRITE_SCOPE, + ), + ), + RoleTemplate( + slug="datasource_reader", + name="Datasource reader", + description="Discover and preview policy-visible datasource states.", + permissions=(CATALOGUE_READ_SCOPE,), + ), +) + + +def _router(context: ModuleContext): + from govoplan_datasources.backend.runtime import configure_runtime + + configure_runtime(registry=context.registry) + from govoplan_datasources.backend.router import router + + return router + + +def _provider(context: ModuleContext) -> SqlDatasourceProvider: + return SqlDatasourceProvider(registry=context.registry) + + +def _tenant_summary(session, tenant_id: str) -> dict[str, int]: + return { + "datasources": ( + session.query(datasource_models.DatasourceRecord) + .filter( + datasource_models.DatasourceRecord.tenant_id == tenant_id, + datasource_models.DatasourceRecord.deleted_at.is_(None), + ) + .count() + ), + "datasource_stages_ready": ( + session.query(datasource_models.DatasourceStageRecord) + .filter( + datasource_models.DatasourceStageRecord.tenant_id == tenant_id, + datasource_models.DatasourceStageRecord.state == "ready", + ) + .count() + ), + "datasource_frozen_states": ( + session.query(datasource_models.DatasourceMaterializationRecord) + .filter( + datasource_models.DatasourceMaterializationRecord.tenant_id + == tenant_id, + datasource_models.DatasourceMaterializationRecord.frozen_at.is_not( + None + ), + ) + .count() + ), + } + + +manifest = ModuleManifest( + id=MODULE_ID, + name=MODULE_NAME, + version=MODULE_VERSION, + dependencies=(), + optional_dependencies=( + "access", + "audit", + "connectors", + "files", + "notifications", + "policy", + ), + optional_capabilities=( + CAPABILITY_AUTH_PRINCIPAL_RESOLVER, + CAPABILITY_AUTH_PERMISSION_EVALUATOR, + CAPABILITY_DATASOURCE_ORIGINS, + ), + provides_interfaces=( + ModuleInterfaceProvider( + name="datasources.catalogue", + version=DATASOURCE_INTERFACE_VERSION, + ), + ModuleInterfaceProvider( + name="datasources.lifecycle", + version=DATASOURCE_INTERFACE_VERSION, + ), + ModuleInterfaceProvider( + name="datasources.materializations", + version=DATASOURCE_INTERFACE_VERSION, + ), + ModuleInterfaceProvider( + name="datasources.staging", + version=DATASOURCE_INTERFACE_VERSION, + ), + ), + requires_interfaces=( + ModuleInterfaceRequirement( + name="connectors.datasource_origins", + version_min="0.1.0", + version_max_exclusive="1.0.0", + optional=True, + ), + ), + permissions=PERMISSIONS, + role_templates=ROLE_TEMPLATES, + nav_items=( + NavItem( + path="/datasources", + label="Datasources", + icon="database-zap", + required_any=(CATALOGUE_READ_SCOPE, ADMIN_SCOPE), + order=70, + ), + ), + frontend=FrontendModule( + module_id=MODULE_ID, + package_name="@govoplan/datasources-webui", + nav_items=( + NavItem( + path="/datasources", + label="Datasources", + icon="database-zap", + required_any=(CATALOGUE_READ_SCOPE, ADMIN_SCOPE), + order=70, + ), + ), + ), + route_factory=_router, + capability_factories={ + CAPABILITY_DATASOURCE_CATALOGUE: _provider, + CAPABILITY_DATASOURCE_LIFECYCLE: _provider, + }, + tenant_summary_providers=(_tenant_summary,), + migration_spec=MigrationSpec( + module_id=MODULE_ID, + metadata=Base.metadata, + script_location=str(Path(__file__).with_name("migrations") / "versions"), + retirement_supported=True, + retirement_provider=drop_table_retirement_provider( + datasource_models.DatasourceStageRecord, + datasource_models.DatasourceMaterializationRecord, + datasource_models.DatasourceRecord, + label="Datasources", + ), + retirement_notes=( + "Destructive retirement drops datasource catalogue entries, immutable " + "materializations, and staging evidence after a database snapshot." + ), + ), + uninstall_guard_providers=( + persistent_table_uninstall_guard( + datasource_models.DatasourceRecord, + datasource_models.DatasourceMaterializationRecord, + datasource_models.DatasourceStageRecord, + label="Datasources", + ), + ), + documentation=( + DocumentationTopic( + id="datasources.lifecycle", + title="Datasource lifecycle", + summary="Governed live, cached, and static data with staging and frozen states.", + body=( + "Datasources is the provider-neutral catalogue consumed by Dataflow, " + "Workflow, Reporting, and policy-aware modules. Static data is staged " + "before promotion. Cached data refreshes connector origins into immutable " + "materializations. Live data is read through a connector and may be frozen " + "for reproducible evidence. Connectors owns protocols and credentials; " + "Datasources owns data identity, provenance, lifecycle, and read semantics." + ), + layer="available", + documentation_types=("admin", "user"), + audience=("operator", "module_admin", "power_user", "product_owner"), + related_modules=( + "connectors", + "dataflow", + "workflow", + "files", + "reporting", + "risk_compliance", + ), + order=70, + ), + ), +) + + +def get_manifest() -> ModuleManifest: + return manifest + + +__all__ = [ + "DATASOURCE_INTERFACE_VERSION", + "MODULE_ID", + "MODULE_VERSION", + "get_manifest", + "manifest", +] diff --git a/src/govoplan_datasources/backend/migrations/__init__.py b/src/govoplan_datasources/backend/migrations/__init__.py new file mode 100644 index 0000000..4b2e6f5 --- /dev/null +++ b/src/govoplan_datasources/backend/migrations/__init__.py @@ -0,0 +1 @@ +"""Datasource Alembic migrations.""" diff --git a/src/govoplan_datasources/backend/migrations/versions/__init__.py b/src/govoplan_datasources/backend/migrations/versions/__init__.py new file mode 100644 index 0000000..53d82ed --- /dev/null +++ b/src/govoplan_datasources/backend/migrations/versions/__init__.py @@ -0,0 +1 @@ +"""Datasource Alembic revisions.""" diff --git a/src/govoplan_datasources/backend/migrations/versions/f3a8d2c7b1e0_v0114_datasources_baseline.py b/src/govoplan_datasources/backend/migrations/versions/f3a8d2c7b1e0_v0114_datasources_baseline.py new file mode 100644 index 0000000..a4e9349 --- /dev/null +++ b/src/govoplan_datasources/backend/migrations/versions/f3a8d2c7b1e0_v0114_datasources_baseline.py @@ -0,0 +1,417 @@ +"""v0.1.14 Datasources baseline + +Revision ID: f3a8d2c7b1e0 +Revises: None +Create Date: 2026-07-28 00:00:00.000000 +""" +from __future__ import annotations + +from alembic import op +import sqlalchemy as sa + + +revision = "f3a8d2c7b1e0" +down_revision = None +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.create_table( + "datasource_catalogue", + sa.Column("id", sa.String(length=36), nullable=False), + sa.Column("tenant_id", sa.String(length=36), nullable=False), + sa.Column("source_name", sa.String(length=120), nullable=False), + sa.Column("name", sa.String(length=300), nullable=False), + sa.Column("description", sa.Text(), nullable=True), + sa.Column("kind", sa.String(length=30), nullable=False), + sa.Column("mode", sa.String(length=20), nullable=False), + sa.Column("shape", sa.String(length=30), nullable=False), + sa.Column("status", sa.String(length=30), nullable=False), + sa.Column("provider", sa.String(length=100), nullable=True), + sa.Column("provider_ref", sa.String(length=500), nullable=True), + sa.Column("schema_version", sa.Integer(), nullable=False), + sa.Column("schema", sa.JSON(), nullable=False), + sa.Column("fingerprint", sa.String(length=64), nullable=False), + sa.Column("current_materialization_id", sa.String(length=36), nullable=True), + sa.Column("row_count", sa.Integer(), nullable=True), + sa.Column("byte_count", sa.Integer(), nullable=True), + sa.Column("provenance", sa.JSON(), nullable=False), + sa.Column("metadata", sa.JSON(), nullable=False), + sa.Column("created_by", sa.String(length=255), nullable=True), + sa.Column("updated_by", sa.String(length=255), nullable=True), + sa.Column("deleted_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), + sa.PrimaryKeyConstraint("id", name=op.f("pk_datasource_catalogue")), + sa.UniqueConstraint( + "tenant_id", + "source_name", + name="uq_datasource_source_name", + ), + ) + op.create_index( + op.f("ix_datasource_catalogue_created_by"), + "datasource_catalogue", + ["created_by"], + unique=False, + ) + op.create_index( + op.f("ix_datasource_catalogue_current_materialization_id"), + "datasource_catalogue", + ["current_materialization_id"], + unique=False, + ) + op.create_index( + op.f("ix_datasource_catalogue_deleted_at"), + "datasource_catalogue", + ["deleted_at"], + unique=False, + ) + op.create_index( + op.f("ix_datasource_catalogue_fingerprint"), + "datasource_catalogue", + ["fingerprint"], + unique=False, + ) + op.create_index( + op.f("ix_datasource_catalogue_kind"), + "datasource_catalogue", + ["kind"], + unique=False, + ) + op.create_index( + op.f("ix_datasource_catalogue_mode"), + "datasource_catalogue", + ["mode"], + unique=False, + ) + op.create_index( + op.f("ix_datasource_catalogue_provider"), + "datasource_catalogue", + ["provider"], + unique=False, + ) + op.create_index( + op.f("ix_datasource_catalogue_shape"), + "datasource_catalogue", + ["shape"], + unique=False, + ) + op.create_index( + op.f("ix_datasource_catalogue_status"), + "datasource_catalogue", + ["status"], + unique=False, + ) + op.create_index( + op.f("ix_datasource_catalogue_tenant_id"), + "datasource_catalogue", + ["tenant_id"], + unique=False, + ) + op.create_index( + op.f("ix_datasource_catalogue_updated_by"), + "datasource_catalogue", + ["updated_by"], + unique=False, + ) + op.create_index( + "ix_datasource_catalogue_origin", + "datasource_catalogue", + ["tenant_id", "provider", "provider_ref"], + unique=False, + ) + op.create_index( + "ix_datasource_catalogue_tenant_mode", + "datasource_catalogue", + ["tenant_id", "mode"], + unique=False, + ) + op.create_index( + "ix_datasource_catalogue_tenant_status", + "datasource_catalogue", + ["tenant_id", "status"], + unique=False, + ) + + op.create_table( + "datasource_materializations", + sa.Column("id", sa.String(length=36), nullable=False), + sa.Column("tenant_id", sa.String(length=36), nullable=False), + sa.Column("datasource_id", sa.String(length=36), nullable=False), + sa.Column("revision", sa.Integer(), nullable=False), + sa.Column("state", sa.String(length=30), nullable=False), + sa.Column("schema_version", sa.Integer(), nullable=False), + sa.Column("schema", sa.JSON(), nullable=False), + sa.Column("rows", sa.JSON(), nullable=False), + sa.Column("fingerprint", sa.String(length=64), nullable=False), + sa.Column("row_count", sa.Integer(), nullable=False), + sa.Column("byte_count", sa.Integer(), nullable=False), + sa.Column("frozen_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("frozen_label", sa.String(length=300), nullable=True), + sa.Column("source_timestamp", sa.DateTime(timezone=True), nullable=True), + sa.Column("provenance", sa.JSON(), nullable=False), + sa.Column("metadata", sa.JSON(), nullable=False), + sa.Column("created_by", sa.String(length=255), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), + sa.ForeignKeyConstraint( + ["datasource_id"], + ["datasource_catalogue.id"], + name=op.f( + "fk_datasource_materializations_datasource_id_datasource_catalogue" + ), + ondelete="CASCADE", + ), + sa.PrimaryKeyConstraint("id", name=op.f("pk_datasource_materializations")), + sa.UniqueConstraint( + "datasource_id", + "revision", + name="uq_datasource_materialization_revision", + ), + ) + op.create_index( + op.f("ix_datasource_materializations_created_by"), + "datasource_materializations", + ["created_by"], + unique=False, + ) + op.create_index( + op.f("ix_datasource_materializations_datasource_id"), + "datasource_materializations", + ["datasource_id"], + unique=False, + ) + op.create_index( + op.f("ix_datasource_materializations_fingerprint"), + "datasource_materializations", + ["fingerprint"], + unique=False, + ) + op.create_index( + op.f("ix_datasource_materializations_state"), + "datasource_materializations", + ["state"], + unique=False, + ) + op.create_index( + op.f("ix_datasource_materializations_tenant_id"), + "datasource_materializations", + ["tenant_id"], + unique=False, + ) + op.create_index( + "ix_datasource_materializations_frozen", + "datasource_materializations", + ["tenant_id", "datasource_id", "frozen_at"], + unique=False, + ) + op.create_index( + "ix_datasource_materializations_tenant_source", + "datasource_materializations", + ["tenant_id", "datasource_id"], + unique=False, + ) + + op.create_table( + "datasource_stages", + sa.Column("id", sa.String(length=36), nullable=False), + sa.Column("tenant_id", sa.String(length=36), nullable=False), + sa.Column("target_datasource_id", sa.String(length=36), nullable=True), + sa.Column("name", sa.String(length=300), nullable=False), + sa.Column("source_name", sa.String(length=120), nullable=False), + sa.Column("description", sa.Text(), nullable=True), + sa.Column("kind", sa.String(length=30), nullable=False), + sa.Column("mode", sa.String(length=20), nullable=False), + sa.Column("shape", sa.String(length=30), nullable=False), + sa.Column("state", sa.String(length=30), nullable=False), + sa.Column("provider", sa.String(length=100), nullable=True), + sa.Column("provider_ref", sa.String(length=500), nullable=True), + sa.Column("schema", sa.JSON(), nullable=False), + sa.Column("rows", sa.JSON(), nullable=False), + sa.Column("fingerprint", sa.String(length=64), nullable=False), + sa.Column("row_count", sa.Integer(), nullable=False), + sa.Column("byte_count", sa.Integer(), nullable=False), + sa.Column("validation", sa.JSON(), nullable=False), + sa.Column("provenance", sa.JSON(), nullable=False), + sa.Column("metadata", sa.JSON(), nullable=False), + sa.Column("promoted_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("promoted_materialization_id", sa.String(length=36), nullable=True), + sa.Column("created_by", sa.String(length=255), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), + sa.ForeignKeyConstraint( + ["target_datasource_id"], + ["datasource_catalogue.id"], + name=op.f("fk_datasource_stages_target_datasource_id_datasource_catalogue"), + ondelete="SET NULL", + ), + sa.PrimaryKeyConstraint("id", name=op.f("pk_datasource_stages")), + ) + op.create_index( + op.f("ix_datasource_stages_created_by"), + "datasource_stages", + ["created_by"], + unique=False, + ) + op.create_index( + op.f("ix_datasource_stages_fingerprint"), + "datasource_stages", + ["fingerprint"], + unique=False, + ) + op.create_index( + op.f("ix_datasource_stages_promoted_materialization_id"), + "datasource_stages", + ["promoted_materialization_id"], + unique=False, + ) + op.create_index( + op.f("ix_datasource_stages_state"), + "datasource_stages", + ["state"], + unique=False, + ) + op.create_index( + op.f("ix_datasource_stages_target_datasource_id"), + "datasource_stages", + ["target_datasource_id"], + unique=False, + ) + op.create_index( + op.f("ix_datasource_stages_tenant_id"), + "datasource_stages", + ["tenant_id"], + unique=False, + ) + op.create_index( + "ix_datasource_stages_target", + "datasource_stages", + ["tenant_id", "target_datasource_id"], + unique=False, + ) + op.create_index( + "ix_datasource_stages_tenant_state", + "datasource_stages", + ["tenant_id", "state"], + unique=False, + ) + + +def downgrade() -> None: + op.drop_index("ix_datasource_stages_tenant_state", table_name="datasource_stages") + op.drop_index("ix_datasource_stages_target", table_name="datasource_stages") + op.drop_index( + op.f("ix_datasource_stages_tenant_id"), + table_name="datasource_stages", + ) + op.drop_index( + op.f("ix_datasource_stages_target_datasource_id"), + table_name="datasource_stages", + ) + op.drop_index( + op.f("ix_datasource_stages_state"), + table_name="datasource_stages", + ) + op.drop_index( + op.f("ix_datasource_stages_promoted_materialization_id"), + table_name="datasource_stages", + ) + op.drop_index( + op.f("ix_datasource_stages_fingerprint"), + table_name="datasource_stages", + ) + op.drop_index( + op.f("ix_datasource_stages_created_by"), + table_name="datasource_stages", + ) + op.drop_table("datasource_stages") + + op.drop_index( + "ix_datasource_materializations_tenant_source", + table_name="datasource_materializations", + ) + op.drop_index( + "ix_datasource_materializations_frozen", + table_name="datasource_materializations", + ) + op.drop_index( + op.f("ix_datasource_materializations_tenant_id"), + table_name="datasource_materializations", + ) + op.drop_index( + op.f("ix_datasource_materializations_state"), + table_name="datasource_materializations", + ) + op.drop_index( + op.f("ix_datasource_materializations_fingerprint"), + table_name="datasource_materializations", + ) + op.drop_index( + op.f("ix_datasource_materializations_datasource_id"), + table_name="datasource_materializations", + ) + op.drop_index( + op.f("ix_datasource_materializations_created_by"), + table_name="datasource_materializations", + ) + op.drop_table("datasource_materializations") + + op.drop_index( + "ix_datasource_catalogue_tenant_status", + table_name="datasource_catalogue", + ) + op.drop_index( + "ix_datasource_catalogue_tenant_mode", + table_name="datasource_catalogue", + ) + op.drop_index( + "ix_datasource_catalogue_origin", + table_name="datasource_catalogue", + ) + op.drop_index( + op.f("ix_datasource_catalogue_updated_by"), + table_name="datasource_catalogue", + ) + op.drop_index( + op.f("ix_datasource_catalogue_tenant_id"), + table_name="datasource_catalogue", + ) + op.drop_index( + op.f("ix_datasource_catalogue_status"), + table_name="datasource_catalogue", + ) + op.drop_index( + op.f("ix_datasource_catalogue_shape"), + table_name="datasource_catalogue", + ) + op.drop_index( + op.f("ix_datasource_catalogue_provider"), + table_name="datasource_catalogue", + ) + op.drop_index( + op.f("ix_datasource_catalogue_mode"), + table_name="datasource_catalogue", + ) + op.drop_index( + op.f("ix_datasource_catalogue_kind"), + table_name="datasource_catalogue", + ) + op.drop_index( + op.f("ix_datasource_catalogue_fingerprint"), + table_name="datasource_catalogue", + ) + op.drop_index( + op.f("ix_datasource_catalogue_deleted_at"), + table_name="datasource_catalogue", + ) + op.drop_index( + op.f("ix_datasource_catalogue_current_materialization_id"), + table_name="datasource_catalogue", + ) + op.drop_index( + op.f("ix_datasource_catalogue_created_by"), + table_name="datasource_catalogue", + ) + op.drop_table("datasource_catalogue") diff --git a/src/govoplan_datasources/backend/router.py b/src/govoplan_datasources/backend/router.py new file mode 100644 index 0000000..0d689d9 --- /dev/null +++ b/src/govoplan_datasources/backend/router.py @@ -0,0 +1,624 @@ +from __future__ import annotations + +from fastapi import APIRouter, Depends, HTTPException, Query, status +from sqlalchemy.orm import Session + +from govoplan_core.audit.logging import audit_event +from govoplan_core.auth import ApiPrincipal, get_api_principal, has_scope +from govoplan_core.core.datasources import ( + DatasourceAccessError, + DatasourceDescriptor, + DatasourceError, + DatasourceMaterialization, + DatasourceNotFoundError, + DatasourceOrigin, + DatasourceReadRequest, + DatasourceStage, + DatasourceStageInput, + DatasourceUnavailableError, + datasource_origins, +) +from govoplan_core.db.session import get_session +from govoplan_datasources.backend.runtime import get_registry +from govoplan_datasources.backend.schemas import ( + DatasourceFieldResponse, + DatasourceFreezeRequest, + DatasourceListResponse, + DatasourceMaterializationListResponse, + DatasourceMaterializationResponse, + DatasourceOriginListResponse, + DatasourceOriginRegisterRequest, + DatasourceOriginResponse, + DatasourcePreviewResponse, + DatasourceResponse, + DatasourceRetireResponse, + DatasourceStageCreateRequest, + DatasourceStageListResponse, + DatasourceStagePromoteRequest, + DatasourceStagePromoteResponse, + DatasourceStageResponse, +) +from govoplan_datasources.backend.service import ( + ADMIN_SCOPE, + CATALOGUE_READ_SCOPE, + SOURCE_WRITE_SCOPE, + STAGE_WRITE_SCOPE, + SqlDatasourceProvider, +) +from govoplan_datasources.backend.tabular import parse_csv_rows + + +router = APIRouter(prefix="/datasources", tags=["datasources"]) + + +def _provider() -> SqlDatasourceProvider: + return SqlDatasourceProvider(registry=get_registry()) + + +def _require_any_scope(principal: ApiPrincipal, *scopes: str) -> None: + if any(has_scope(principal, scope) for scope in scopes): + return + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail=f"Missing one of the required scopes: {', '.join(scopes)}", + ) + + +def _http_error(exc: DatasourceError) -> HTTPException: + if isinstance(exc, DatasourceNotFoundError): + return HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) + if isinstance(exc, DatasourceAccessError): + return HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=str(exc)) + if isinstance(exc, DatasourceUnavailableError): + return HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(exc)) + return HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, + detail=str(exc), + ) + + +@router.get("/origins", response_model=DatasourceOriginListResponse) +def api_list_origins( + query: str = Query(default="", max_length=200), + limit: int = Query(default=100, ge=1, le=100), + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> DatasourceOriginListResponse: + _require_any_scope( + principal, + CATALOGUE_READ_SCOPE, + SOURCE_WRITE_SCOPE, + ADMIN_SCOPE, + ) + provider = datasource_origins(get_registry()) + if provider is None: + return DatasourceOriginListResponse(available=False, origins=[]) + try: + origins = provider.list_origins( + session, + principal, + query=query, + limit=limit, + ) + except DatasourceError as exc: + raise _http_error(exc) from exc + return DatasourceOriginListResponse( + available=True, + origins=[_origin_response(origin) for origin in origins], + ) + + +@router.post( + "/origins/register", + response_model=DatasourceResponse, + status_code=status.HTTP_201_CREATED, +) +def api_register_origin( + payload: DatasourceOriginRegisterRequest, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> DatasourceResponse: + _require_any_scope(principal, SOURCE_WRITE_SCOPE, ADMIN_SCOPE) + try: + datasource = _provider().register_origin( + session, + principal, + origin_ref=payload.origin_ref, + name=payload.name, + source_name=payload.source_name, + mode=payload.mode, + description=payload.description, + ) + except DatasourceError as exc: + raise _http_error(exc) from exc + _audit( + session, + principal, + action="datasources.origin.registered", + object_type="datasource", + object_id=datasource.ref, + details={ + "origin_ref": payload.origin_ref, + "mode": datasource.mode, + "source_name": datasource.source_name, + }, + ) + session.commit() + return _datasource_response(datasource) + + +@router.get("/stages", response_model=DatasourceStageListResponse) +def api_list_stages( + limit: int = Query(default=100, ge=1, le=100), + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> DatasourceStageListResponse: + _require_any_scope( + principal, + CATALOGUE_READ_SCOPE, + STAGE_WRITE_SCOPE, + ADMIN_SCOPE, + ) + try: + stages = _provider().list_stages(session, principal, limit=limit) + except DatasourceError as exc: + raise _http_error(exc) from exc + return DatasourceStageListResponse( + stages=[_stage_response(stage) for stage in stages] + ) + + +@router.post( + "/stages", + response_model=DatasourceStageResponse, + status_code=status.HTTP_201_CREATED, +) +def api_create_stage( + payload: DatasourceStageCreateRequest, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> DatasourceStageResponse: + _require_any_scope(principal, STAGE_WRITE_SCOPE, ADMIN_SCOPE) + try: + rows = ( + tuple(payload.rows or ()) + if payload.format == "json" + else parse_csv_rows(payload.csv_text or "", delimiter=payload.delimiter) + ) + stage = _provider().create_stage( + session, + principal, + stage=DatasourceStageInput( + name=payload.name, + source_name=payload.source_name, + description=payload.description, + kind=payload.kind, + mode=payload.mode, + shape=payload.shape, + rows=rows, + target_datasource_ref=payload.target_datasource_ref, + provider="datasources.upload", + provenance={ + **payload.provenance, + "created_via": "datasources", + "source_format": payload.format, + }, + metadata=payload.metadata, + ), + ) + except DatasourceError as exc: + raise _http_error(exc) from exc + _audit( + session, + principal, + action="datasources.stage.created", + object_type="datasource_stage", + object_id=stage.ref, + details={ + "source_name": stage.source_name, + "mode": stage.mode, + "row_count": stage.row_count, + "fingerprint": stage.fingerprint, + }, + ) + session.commit() + return _stage_response(stage) + + +@router.post( + "/stages/{stage_id}/promote", + response_model=DatasourceStagePromoteResponse, +) +def api_promote_stage( + stage_id: str, + payload: DatasourceStagePromoteRequest, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> DatasourceStagePromoteResponse: + _require_any_scope(principal, STAGE_WRITE_SCOPE, ADMIN_SCOPE) + try: + datasource, materialization = _provider().promote_stage( + session, + principal, + stage_ref=f"stage:{stage_id}", + freeze=payload.freeze, + frozen_label=payload.frozen_label, + ) + except DatasourceError as exc: + raise _http_error(exc) from exc + _audit( + session, + principal, + action="datasources.stage.promoted", + object_type="datasource", + object_id=datasource.ref, + details={ + "stage_ref": f"stage:{stage_id}", + "materialization_ref": materialization.ref, + "revision": materialization.revision, + "frozen": materialization.frozen_at is not None, + }, + ) + session.commit() + return DatasourceStagePromoteResponse( + datasource=_datasource_response(datasource), + materialization=_materialization_response(materialization), + ) + + +@router.get("", response_model=DatasourceListResponse) +def api_list_datasources( + query: str = Query(default="", max_length=200), + limit: int = Query(default=100, ge=1, le=100), + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> DatasourceListResponse: + _require_any_scope(principal, CATALOGUE_READ_SCOPE, ADMIN_SCOPE) + try: + items = _provider().list_datasources( + session, + principal, + query=query, + limit=limit, + ) + except DatasourceError as exc: + raise _http_error(exc) from exc + return DatasourceListResponse( + datasources=[_datasource_response(item) for item in items] + ) + + +@router.get("/{datasource_id}", response_model=DatasourceResponse) +def api_get_datasource( + datasource_id: str, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> DatasourceResponse: + _require_any_scope(principal, CATALOGUE_READ_SCOPE, ADMIN_SCOPE) + try: + item = _provider().get_datasource( + session, + principal, + datasource_ref=f"datasource:{datasource_id}", + ) + except DatasourceError as exc: + raise _http_error(exc) from exc + if item is None: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail="Datasource not found.", + ) + return _datasource_response(item) + + +@router.get( + "/{datasource_id}/preview", + response_model=DatasourcePreviewResponse, +) +def api_preview_datasource( + datasource_id: str, + limit: int = Query(default=100, ge=1, le=500), + offset: int = Query(default=0, ge=0), + consistency: str = Query(default="current", pattern="^(current|live|frozen)$"), + materialization_ref: str | None = Query(default=None, max_length=120), + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> DatasourcePreviewResponse: + _require_any_scope(principal, CATALOGUE_READ_SCOPE, ADMIN_SCOPE) + try: + result = _provider().read_datasource( + session, + principal, + request=DatasourceReadRequest( + datasource_ref=f"datasource:{datasource_id}", + materialization_ref=materialization_ref, + consistency=consistency, # type: ignore[arg-type] + limit=limit, + offset=offset, + ), + ) + except DatasourceError as exc: + raise _http_error(exc) from exc + return DatasourcePreviewResponse( + datasource=_datasource_response(result.datasource), + rows=[dict(row) for row in result.rows], + total_rows=result.total_rows, + truncated=result.truncated, + materialization=( + _materialization_response(result.materialization) + if result.materialization + else None + ), + ) + + +@router.get( + "/{datasource_id}/materializations", + response_model=DatasourceMaterializationListResponse, +) +def api_list_materializations( + datasource_id: str, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> DatasourceMaterializationListResponse: + _require_any_scope(principal, CATALOGUE_READ_SCOPE, ADMIN_SCOPE) + try: + rows = _provider().list_materializations( + session, + principal, + datasource_ref=f"datasource:{datasource_id}", + ) + except DatasourceError as exc: + raise _http_error(exc) from exc + return DatasourceMaterializationListResponse( + materializations=[_materialization_response(item) for item in rows] + ) + + +@router.post( + "/{datasource_id}/refresh", + response_model=DatasourceStagePromoteResponse, +) +def api_refresh_datasource( + datasource_id: str, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> DatasourceStagePromoteResponse: + _require_any_scope(principal, SOURCE_WRITE_SCOPE, ADMIN_SCOPE) + try: + datasource, materialization = _provider().refresh_datasource( + session, + principal, + datasource_ref=f"datasource:{datasource_id}", + ) + except DatasourceError as exc: + raise _http_error(exc) from exc + _audit( + session, + principal, + action="datasources.refreshed", + object_type="datasource", + object_id=datasource.ref, + details={ + "materialization_ref": materialization.ref, + "revision": materialization.revision, + "fingerprint": materialization.fingerprint, + }, + ) + session.commit() + return DatasourceStagePromoteResponse( + datasource=_datasource_response(datasource), + materialization=_materialization_response(materialization), + ) + + +@router.post( + "/{datasource_id}/freeze", + response_model=DatasourceMaterializationResponse, +) +def api_freeze_datasource( + datasource_id: str, + payload: DatasourceFreezeRequest, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> DatasourceMaterializationResponse: + _require_any_scope(principal, SOURCE_WRITE_SCOPE, ADMIN_SCOPE) + datasource_ref = f"datasource:{datasource_id}" + try: + materialization = _provider().freeze_datasource( + session, + principal, + datasource_ref=datasource_ref, + label=payload.label, + ) + except DatasourceError as exc: + raise _http_error(exc) from exc + _audit( + session, + principal, + action="datasources.frozen", + object_type="datasource", + object_id=datasource_ref, + details={ + "materialization_ref": materialization.ref, + "revision": materialization.revision, + "label": materialization.frozen_label, + }, + ) + session.commit() + return _materialization_response(materialization) + + +@router.delete( + "/{datasource_id}", + response_model=DatasourceRetireResponse, +) +def api_retire_datasource( + datasource_id: str, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> DatasourceRetireResponse: + _require_any_scope(principal, SOURCE_WRITE_SCOPE, ADMIN_SCOPE) + datasource_ref = f"datasource:{datasource_id}" + try: + _provider().retire_datasource( + session, + principal, + datasource_ref=datasource_ref, + ) + except DatasourceError as exc: + raise _http_error(exc) from exc + _audit( + session, + principal, + action="datasources.retired", + object_type="datasource", + object_id=datasource_ref, + details={}, + ) + session.commit() + return DatasourceRetireResponse(retired=True, datasource_ref=datasource_ref) + + +def _datasource_response(item: DatasourceDescriptor) -> DatasourceResponse: + return DatasourceResponse( + ref=item.ref, + source_name=item.source_name, + name=item.name, + description=item.description, + kind=item.kind, + mode=item.mode, + shape=item.shape, + status=item.status, + provider=item.provider, + provider_ref=item.provider_ref, + schema=[ + DatasourceFieldResponse( + name=field.name, + data_type=field.data_type, + nullable=field.nullable, + ) + for field in item.schema + ], + schema_version=item.schema_version, + fingerprint=item.fingerprint, + current_materialization_ref=item.current_materialization_ref, + row_count=item.row_count, + byte_count=item.byte_count, + updated_at=item.updated_at.isoformat() if item.updated_at else None, + capabilities=list(item.capabilities), + provenance=dict(item.provenance), + metadata=dict(item.metadata), + ) + + +def _materialization_response( + item: DatasourceMaterialization, +) -> DatasourceMaterializationResponse: + return DatasourceMaterializationResponse( + ref=item.ref, + datasource_ref=item.datasource_ref, + revision=item.revision, + state=item.state, + fingerprint=item.fingerprint, + schema=[ + DatasourceFieldResponse( + name=field.name, + data_type=field.data_type, + nullable=field.nullable, + ) + for field in item.schema + ], + row_count=item.row_count, + byte_count=item.byte_count, + frozen_at=item.frozen_at.isoformat() if item.frozen_at else None, + frozen_label=item.frozen_label, + source_timestamp=( + item.source_timestamp.isoformat() if item.source_timestamp else None + ), + created_at=item.created_at.isoformat() if item.created_at else None, + provenance=dict(item.provenance), + metadata=dict(item.metadata), + ) + + +def _stage_response(item: DatasourceStage) -> DatasourceStageResponse: + return DatasourceStageResponse( + ref=item.ref, + name=item.name, + source_name=item.source_name, + kind=item.kind, + mode=item.mode, + shape=item.shape, + state=item.state, + target_datasource_ref=item.target_datasource_ref, + fingerprint=item.fingerprint, + schema=[ + DatasourceFieldResponse( + name=field.name, + data_type=field.data_type, + nullable=field.nullable, + ) + for field in item.schema + ], + row_count=item.row_count, + byte_count=item.byte_count, + validation=dict(item.validation), + created_at=item.created_at.isoformat() if item.created_at else None, + promoted_at=item.promoted_at.isoformat() if item.promoted_at else None, + promoted_materialization_ref=item.promoted_materialization_ref, + provenance=dict(item.provenance), + metadata=dict(item.metadata), + ) + + +def _origin_response(item: DatasourceOrigin) -> DatasourceOriginResponse: + return DatasourceOriginResponse( + ref=item.ref, + source_name=item.source_name, + name=item.name, + description=item.description, + kind=item.kind, + shape=item.shape, + supported_modes=list(item.supported_modes), + provider=item.provider, + schema=[ + DatasourceFieldResponse( + name=field.name, + data_type=field.data_type, + nullable=field.nullable, + ) + for field in item.schema + ], + schema_version=item.schema_version, + fingerprint=item.fingerprint, + row_count=item.row_count, + byte_count=item.byte_count, + updated_at=item.updated_at.isoformat() if item.updated_at else None, + capabilities=list(item.capabilities), + metadata=dict(item.metadata), + ) + + +def _audit( + session: Session, + principal: ApiPrincipal, + *, + action: str, + object_type: str, + object_id: str, + details: dict[str, object], +) -> None: + audit_event( + session, + tenant_id=principal.tenant_id, + user_id=getattr(principal.user, "id", None), + api_key_id=principal.api_key_id, + action=action, + object_type=object_type, + object_id=object_id, + details=details, + ) + + +__all__ = ["router"] diff --git a/src/govoplan_datasources/backend/runtime.py b/src/govoplan_datasources/backend/runtime.py new file mode 100644 index 0000000..aac99c8 --- /dev/null +++ b/src/govoplan_datasources/backend/runtime.py @@ -0,0 +1,16 @@ +from __future__ import annotations + + +_registry: object | None = None + + +def configure_runtime(*, registry: object) -> None: + global _registry + _registry = registry + + +def get_registry() -> object | None: + return _registry + + +__all__ = ["configure_runtime", "get_registry"] diff --git a/src/govoplan_datasources/backend/schemas.py b/src/govoplan_datasources/backend/schemas.py new file mode 100644 index 0000000..4361fe3 --- /dev/null +++ b/src/govoplan_datasources/backend/schemas.py @@ -0,0 +1,181 @@ +from __future__ import annotations + +from typing import Any, Literal + +from pydantic import BaseModel, Field, model_validator + + +DatasourceModeValue = Literal["live", "cached", "static"] +DatasourceKindValue = Literal[ + "upload", + "database", + "http", + "rest", + "directory", + "file", + "feed", + "custom", +] +DatasourceShapeValue = Literal["tabular", "document", "binary", "directory", "stream"] + + +class DatasourceFieldResponse(BaseModel): + name: str + data_type: str + nullable: bool + + +class DatasourceResponse(BaseModel): + ref: str + source_name: str + name: str + description: str | None + kind: DatasourceKindValue + mode: DatasourceModeValue + shape: DatasourceShapeValue + status: str + provider: str | None + provider_ref: str | None + fields: list[DatasourceFieldResponse] = Field(alias="schema") + schema_version: str + fingerprint: str + current_materialization_ref: str | None + row_count: int | None + byte_count: int | None + updated_at: str | None + capabilities: list[str] + provenance: dict[str, Any] + metadata: dict[str, Any] + + +class DatasourceListResponse(BaseModel): + datasources: list[DatasourceResponse] + + +class DatasourceMaterializationResponse(BaseModel): + ref: str + datasource_ref: str + revision: int + state: str + fingerprint: str + fields: list[DatasourceFieldResponse] = Field(alias="schema") + row_count: int | None + byte_count: int | None + frozen_at: str | None + frozen_label: str | None + source_timestamp: str | None + created_at: str | None + provenance: dict[str, Any] + metadata: dict[str, Any] + + +class DatasourceMaterializationListResponse(BaseModel): + materializations: list[DatasourceMaterializationResponse] + + +class DatasourceStageResponse(BaseModel): + ref: str + name: str + source_name: str + kind: DatasourceKindValue + mode: DatasourceModeValue + shape: DatasourceShapeValue + state: str + target_datasource_ref: str | None + fingerprint: str + fields: list[DatasourceFieldResponse] = Field(alias="schema") + row_count: int | None + byte_count: int | None + validation: dict[str, Any] + created_at: str | None + promoted_at: str | None + promoted_materialization_ref: str | None + provenance: dict[str, Any] + metadata: dict[str, Any] + + +class DatasourceStageListResponse(BaseModel): + stages: list[DatasourceStageResponse] + + +class DatasourceStageCreateRequest(BaseModel): + name: str = Field(min_length=1, max_length=300) + source_name: str = Field(min_length=1, max_length=120) + description: str | None = Field(default=None, max_length=4_000) + kind: DatasourceKindValue = "upload" + mode: Literal["static", "cached"] = "static" + shape: Literal["tabular"] = "tabular" + format: Literal["json", "csv"] = "json" + rows: list[dict[str, Any]] | None = Field(default=None, max_length=10_000) + csv_text: str | None = Field(default=None, max_length=5_000_000) + delimiter: Literal[",", ";", "\t", "|"] = "," + target_datasource_ref: str | None = None + provenance: dict[str, Any] = Field(default_factory=dict) + metadata: dict[str, Any] = Field(default_factory=dict) + + @model_validator(mode="after") + def matching_payload(self) -> "DatasourceStageCreateRequest": + if self.format == "json" and self.rows is None: + raise ValueError("JSON stages require rows.") + if self.format == "csv" and not self.csv_text: + raise ValueError("CSV stages require csv_text.") + return self + + +class DatasourceStagePromoteRequest(BaseModel): + freeze: bool = False + frozen_label: str | None = Field(default=None, max_length=300) + + +class DatasourceStagePromoteResponse(BaseModel): + datasource: DatasourceResponse + materialization: DatasourceMaterializationResponse + + +class DatasourceOriginResponse(BaseModel): + ref: str + source_name: str + name: str + description: str | None + kind: DatasourceKindValue + shape: DatasourceShapeValue + supported_modes: list[DatasourceModeValue] + provider: str + fields: list[DatasourceFieldResponse] = Field(alias="schema") + schema_version: str + fingerprint: str + row_count: int | None + byte_count: int | None + updated_at: str | None + capabilities: list[str] + metadata: dict[str, Any] + + +class DatasourceOriginListResponse(BaseModel): + available: bool + origins: list[DatasourceOriginResponse] + + +class DatasourceOriginRegisterRequest(BaseModel): + origin_ref: str = Field(min_length=1, max_length=500) + name: str = Field(min_length=1, max_length=300) + source_name: str = Field(min_length=1, max_length=120) + mode: Literal["live", "cached"] + description: str | None = Field(default=None, max_length=4_000) + + +class DatasourcePreviewResponse(BaseModel): + datasource: DatasourceResponse + rows: list[dict[str, Any]] + total_rows: int + truncated: bool + materialization: DatasourceMaterializationResponse | None + + +class DatasourceFreezeRequest(BaseModel): + label: str | None = Field(default=None, max_length=300) + + +class DatasourceRetireResponse(BaseModel): + retired: bool + datasource_ref: str diff --git a/src/govoplan_datasources/backend/service.py b/src/govoplan_datasources/backend/service.py new file mode 100644 index 0000000..0752211 --- /dev/null +++ b/src/govoplan_datasources/backend/service.py @@ -0,0 +1,1111 @@ +from __future__ import annotations + +from collections.abc import Mapping, Sequence +from dataclasses import replace +from typing import Any, cast + +from sqlalchemy import func, or_, select +from sqlalchemy.orm import Session + +from govoplan_core.auth import ApiPrincipal, has_scope +from govoplan_core.core.datasources import ( + DatasourceAccessError, + DatasourceDescriptor, + DatasourceError, + DatasourceField, + DatasourceMaterialization, + DatasourceMode, + DatasourceNotFoundError, + DatasourceOrigin, + DatasourceOriginReadRequest, + DatasourceReadRequest, + DatasourceReadResult, + DatasourceStage, + DatasourceStageInput, + DatasourceUnavailableError, + DatasourceValidationError, + datasource_origins, +) +from govoplan_core.db.base import utcnow +from govoplan_datasources.backend.db.models import ( + DatasourceMaterializationRecord, + DatasourceRecord, + DatasourceStageRecord, +) +from govoplan_datasources.backend.tabular import ( + MAX_READ_ROWS, + MAX_STAGE_ROWS, + encoded_size, + field_payload, + fingerprint_rows, + infer_schema, + normalize_rows, +) + + +CATALOGUE_READ_SCOPE = "datasources:catalogue:read" +SOURCE_WRITE_SCOPE = "datasources:source:write" +STAGE_WRITE_SCOPE = "datasources:stage:write" +ADMIN_SCOPE = "datasources:source:admin" + + +class SqlDatasourceProvider: + def __init__(self, *, registry: object | None = None) -> None: + self._registry = registry + + def list_datasources( + self, + session: object, + principal: object, + *, + query: str = "", + limit: int = 100, + ) -> Sequence[DatasourceDescriptor]: + db, api_principal = _context(session, principal, CATALOGUE_READ_SCOPE) + statement = ( + select(DatasourceRecord) + .where( + DatasourceRecord.tenant_id == api_principal.tenant_id, + DatasourceRecord.deleted_at.is_(None), + ) + .order_by(DatasourceRecord.name.asc(), DatasourceRecord.id.asc()) + .limit(max(1, min(int(limit), 100))) + ) + cleaned_query = query.strip() + if cleaned_query: + pattern = f"%{_escape_like(cleaned_query)}%" + statement = statement.where( + or_( + DatasourceRecord.name.ilike(pattern, escape="\\"), + DatasourceRecord.source_name.ilike(pattern, escape="\\"), + DatasourceRecord.description.ilike(pattern, escape="\\"), + ) + ) + return tuple(_datasource_dto(item) for item in db.scalars(statement)) + + def get_datasource( + self, + session: object, + principal: object, + *, + datasource_ref: str, + ) -> DatasourceDescriptor | None: + db, api_principal = _context(session, principal, CATALOGUE_READ_SCOPE) + item = _datasource_record( + db, + tenant_id=api_principal.tenant_id, + datasource_ref=datasource_ref, + ) + return _datasource_dto(item) if item is not None else None + + def read_datasource( + self, + session: object, + principal: object, + *, + request: DatasourceReadRequest, + ) -> DatasourceReadResult: + db, api_principal = _context(session, principal, CATALOGUE_READ_SCOPE) + item = _required_datasource( + db, + tenant_id=api_principal.tenant_id, + datasource_ref=request.datasource_ref, + ) + limit = max(1, min(int(request.limit), MAX_READ_ROWS)) + offset = max(0, int(request.offset)) + columns = tuple(dict.fromkeys(request.columns)) + + direct_live = request.consistency == "live" + implicit_live = ( + item.mode == "live" + and not request.materialization_ref + and request.consistency == "current" + ) + if direct_live or implicit_live: + if not item.provider_ref: + raise DatasourceUnavailableError( + "This datasource has no live origin." + ) + result = self._read_live( + db, + api_principal, + item=item, + request=request, + limit=limit, + offset=offset, + columns=columns, + ) + if ( + request.expected_fingerprint + and request.expected_fingerprint != result.datasource.fingerprint + ): + raise DatasourceValidationError( + "The live datasource fingerprint changed; refresh the consuming definition." + ) + return result + + materialization = _selected_materialization( + db, + item=item, + request=request, + ) + if materialization is None: + message = ( + "No frozen state is available for this datasource." + if request.consistency == "frozen" + else "This datasource has no materialized state." + ) + raise DatasourceUnavailableError(message) + if ( + request.expected_fingerprint + and request.expected_fingerprint != materialization.fingerprint + ): + raise DatasourceValidationError( + "The datasource fingerprint changed; refresh the consuming definition." + ) + _validate_columns(materialization.schema_, columns) + window = materialization.rows[offset : offset + limit] + rows = tuple(_select_columns(row, columns) for row in window) + descriptor = _datasource_dto(item) + return DatasourceReadResult( + datasource=replace( + descriptor, + schema=_fields(materialization.schema_), + schema_version=str(materialization.schema_version), + fingerprint=materialization.fingerprint, + row_count=materialization.row_count, + byte_count=materialization.byte_count, + ), + rows=rows, + total_rows=materialization.row_count, + truncated=offset + len(rows) < materialization.row_count, + materialization=_materialization_dto(materialization), + ) + + def list_materializations( + self, + session: object, + principal: object, + *, + datasource_ref: str, + ) -> Sequence[DatasourceMaterialization]: + db, api_principal = _context(session, principal, CATALOGUE_READ_SCOPE) + item = _required_datasource( + db, + tenant_id=api_principal.tenant_id, + datasource_ref=datasource_ref, + ) + statement = ( + select(DatasourceMaterializationRecord) + .where( + DatasourceMaterializationRecord.tenant_id == api_principal.tenant_id, + DatasourceMaterializationRecord.datasource_id == item.id, + ) + .order_by(DatasourceMaterializationRecord.revision.desc()) + ) + return tuple(_materialization_dto(row) for row in db.scalars(statement)) + + def list_stages( + self, + session: object, + principal: object, + *, + limit: int = 100, + ) -> Sequence[DatasourceStage]: + db, api_principal = _context( + session, + principal, + (CATALOGUE_READ_SCOPE, STAGE_WRITE_SCOPE), + ) + statement = ( + select(DatasourceStageRecord) + .where(DatasourceStageRecord.tenant_id == api_principal.tenant_id) + .order_by(DatasourceStageRecord.created_at.desc()) + .limit(max(1, min(int(limit), 100))) + ) + return tuple(_stage_dto(row) for row in db.scalars(statement)) + + def create_stage( + self, + session: object, + principal: object, + *, + stage: DatasourceStageInput, + ) -> DatasourceStage: + db, api_principal = _context(session, principal, STAGE_WRITE_SCOPE) + name = stage.name.strip() + source_name = stage.source_name.strip() + if not name: + raise DatasourceValidationError("Stage name is required.") + if not source_name: + raise DatasourceValidationError("Datasource key is required.") + if stage.shape != "tabular": + raise DatasourceValidationError( + "The first staging provider supports tabular data only." + ) + if stage.mode not in {"static", "cached"}: + raise DatasourceValidationError( + "Uploaded stages can be promoted as static or cached data." + ) + target = None + if stage.target_datasource_ref: + target = _required_datasource( + db, + tenant_id=api_principal.tenant_id, + datasource_ref=stage.target_datasource_ref, + ) + if target.shape != stage.shape or target.mode != stage.mode: + raise DatasourceValidationError( + "A stage can only update a datasource with the same mode and shape." + ) + rows = normalize_rows(stage.rows) + schema = infer_schema(rows) + fingerprint = fingerprint_rows(rows, schema) + item = DatasourceStageRecord( + tenant_id=api_principal.tenant_id, + target_datasource_id=target.id if target else None, + name=name, + source_name=source_name, + description=_clean_optional(stage.description), + kind=stage.kind, + mode=stage.mode, + shape=stage.shape, + state="ready", + provider=_clean_optional(stage.provider), + provider_ref=_clean_optional(stage.provider_ref), + schema_=[field_payload(field) for field in schema], + rows=list(rows), + fingerprint=fingerprint, + row_count=len(rows), + byte_count=encoded_size(rows), + validation_={"valid": True, "errors": [], "warnings": []}, + provenance_=dict(stage.provenance), + metadata_=dict(stage.metadata), + created_by=_actor_id(api_principal), + ) + db.add(item) + db.flush() + return _stage_dto(item) + + def promote_stage( + self, + session: object, + principal: object, + *, + stage_ref: str, + freeze: bool = False, + frozen_label: str | None = None, + ) -> tuple[DatasourceDescriptor, DatasourceMaterialization]: + db, api_principal = _context(session, principal, STAGE_WRITE_SCOPE) + stage = _required_stage( + db, + tenant_id=api_principal.tenant_id, + stage_ref=stage_ref, + ) + if stage.state != "ready": + raise DatasourceValidationError("Only ready stages can be promoted.") + datasource = ( + db.get(DatasourceRecord, stage.target_datasource_id) + if stage.target_datasource_id + else None + ) + if datasource is not None and ( + datasource.tenant_id != api_principal.tenant_id + or datasource.deleted_at is not None + ): + datasource = None + if datasource is None: + _ensure_source_name_available( + db, + tenant_id=api_principal.tenant_id, + source_name=stage.source_name, + ) + datasource = DatasourceRecord( + tenant_id=api_principal.tenant_id, + source_name=stage.source_name, + name=stage.name, + description=stage.description, + kind=stage.kind, + mode=stage.mode, + shape=stage.shape, + status="active", + provider=stage.provider or "datasources.stage", + provider_ref=stage.provider_ref, + schema_version=1, + schema_=list(stage.schema_), + fingerprint=stage.fingerprint, + row_count=stage.row_count, + byte_count=stage.byte_count, + provenance_=dict(stage.provenance_), + metadata_=dict(stage.metadata_), + created_by=_actor_id(api_principal), + updated_by=_actor_id(api_principal), + ) + db.add(datasource) + db.flush() + stage.target_datasource_id = datasource.id + elif datasource.mode != stage.mode or datasource.shape != stage.shape: + raise DatasourceValidationError( + "A stage can only update a datasource with the same mode and shape." + ) + + materialization = _append_materialization( + db, + datasource=datasource, + rows=stage.rows, + schema=stage.schema_, + fingerprint=stage.fingerprint, + byte_count=stage.byte_count, + actor_id=_actor_id(api_principal), + frozen=freeze, + frozen_label=frozen_label, + provenance={ + **dict(stage.provenance_), + "stage_ref": _stage_ref(stage.id), + }, + metadata=dict(stage.metadata_), + set_current=True, + ) + stage.state = "promoted" + stage.promoted_at = utcnow() + stage.promoted_materialization_id = materialization.id + db.flush() + return _datasource_dto(datasource), _materialization_dto(materialization) + + def register_origin( + self, + session: object, + principal: object, + *, + origin_ref: str, + name: str, + source_name: str, + mode: DatasourceMode, + description: str | None = None, + ) -> DatasourceDescriptor: + db, api_principal = _context(session, principal, SOURCE_WRITE_SCOPE) + if mode not in {"live", "cached"}: + raise DatasourceValidationError( + "Connector origins can be registered as live or cached datasources." + ) + origin = self._required_origin(db, api_principal, origin_ref) + if mode not in origin.supported_modes: + raise DatasourceValidationError( + f"The selected origin does not support {mode!r} datasources." + ) + cleaned_name = name.strip() + cleaned_source_name = source_name.strip() + if not cleaned_name or not cleaned_source_name: + raise DatasourceValidationError( + "Datasource name and key are required." + ) + _ensure_source_name_available( + db, + tenant_id=api_principal.tenant_id, + source_name=cleaned_source_name, + ) + item = DatasourceRecord( + tenant_id=api_principal.tenant_id, + source_name=cleaned_source_name, + name=cleaned_name, + description=_clean_optional(description) or origin.description, + kind=origin.kind, + mode=mode, + shape=origin.shape, + status="active", + provider=origin.provider, + provider_ref=origin.ref, + schema_version=_int_schema_version(origin.schema_version), + schema_=[field_payload(field) for field in origin.schema], + fingerprint=origin.fingerprint, + row_count=origin.row_count, + byte_count=origin.byte_count, + provenance_={ + "origin_ref": origin.ref, + "origin_provider": origin.provider, + "registered_at": utcnow().isoformat(), + }, + metadata_=dict(origin.metadata), + created_by=_actor_id(api_principal), + updated_by=_actor_id(api_principal), + ) + db.add(item) + db.flush() + if mode == "cached": + rows, refreshed_origin = self._read_origin_all( + db, + api_principal, + origin_ref=origin_ref, + ) + self._materialize_origin( + db, + item=item, + origin=refreshed_origin, + rows=rows, + actor_id=_actor_id(api_principal), + set_current=True, + ) + return _datasource_dto(item) + + def refresh_datasource( + self, + session: object, + principal: object, + *, + datasource_ref: str, + ) -> tuple[DatasourceDescriptor, DatasourceMaterialization]: + db, api_principal = _context(session, principal, SOURCE_WRITE_SCOPE) + item = _required_datasource( + db, + tenant_id=api_principal.tenant_id, + datasource_ref=datasource_ref, + ) + if item.mode != "cached" or not item.provider_ref: + raise DatasourceValidationError( + "Only cached connector-backed datasources can be refreshed." + ) + rows, origin = self._read_origin_all( + db, + api_principal, + origin_ref=item.provider_ref, + ) + materialization = self._materialize_origin( + db, + item=item, + origin=origin, + rows=rows, + actor_id=_actor_id(api_principal), + set_current=True, + ) + return _datasource_dto(item), _materialization_dto(materialization) + + def freeze_datasource( + self, + session: object, + principal: object, + *, + datasource_ref: str, + label: str | None = None, + ) -> DatasourceMaterialization: + db, api_principal = _context(session, principal, SOURCE_WRITE_SCOPE) + item = _required_datasource( + db, + tenant_id=api_principal.tenant_id, + datasource_ref=datasource_ref, + ) + if item.mode == "live": + if not item.provider_ref: + raise DatasourceUnavailableError( + "The live datasource no longer references an origin." + ) + rows, origin = self._read_origin_all( + db, + api_principal, + origin_ref=item.provider_ref, + ) + materialization = self._materialize_origin( + db, + item=item, + origin=origin, + rows=rows, + actor_id=_actor_id(api_principal), + frozen=True, + frozen_label=label, + set_current=False, + ) + return _materialization_dto(materialization) + + current = _current_materialization(db, item) + if current is None: + raise DatasourceUnavailableError( + "The datasource has no current state to freeze." + ) + materialization = _append_materialization( + db, + datasource=item, + rows=current.rows, + schema=current.schema_, + fingerprint=current.fingerprint, + byte_count=current.byte_count, + actor_id=_actor_id(api_principal), + frozen=True, + frozen_label=label, + source_timestamp=current.source_timestamp, + provenance={ + **dict(current.provenance_), + "frozen_from": _materialization_ref(current.id), + }, + metadata=dict(current.metadata_), + set_current=False, + ) + return _materialization_dto(materialization) + + def retire_datasource( + self, + session: object, + principal: object, + *, + datasource_ref: str, + ) -> DatasourceDescriptor: + db, api_principal = _context(session, principal, SOURCE_WRITE_SCOPE) + item = _required_datasource( + db, + tenant_id=api_principal.tenant_id, + datasource_ref=datasource_ref, + ) + item.status = "retired" + item.deleted_at = utcnow() + item.updated_by = _actor_id(api_principal) + db.flush() + return _datasource_dto(item) + + def _read_live( + self, + session: Session, + principal: ApiPrincipal, + *, + item: DatasourceRecord, + request: DatasourceReadRequest, + limit: int, + offset: int, + columns: tuple[str, ...], + ) -> DatasourceReadResult: + if not item.provider_ref: + raise DatasourceUnavailableError( + "The live datasource no longer references an origin." + ) + provider = datasource_origins(self._registry) + if provider is None: + raise DatasourceUnavailableError( + "No connector origin provider is currently available." + ) + try: + result = provider.read_origin( + session, + principal, + request=DatasourceOriginReadRequest( + origin_ref=item.provider_ref, + limit=limit, + offset=offset, + columns=columns, + expected_fingerprint=request.expected_fingerprint, + ), + ) + except DatasourceError: + raise + except Exception as exc: + raise DatasourceUnavailableError( + f"The datasource origin could not be read: {exc}" + ) from exc + descriptor = replace( + _datasource_dto(item), + schema=result.origin.schema, + schema_version=result.origin.schema_version, + fingerprint=result.origin.fingerprint, + row_count=result.origin.row_count, + byte_count=result.origin.byte_count, + updated_at=result.origin.updated_at, + ) + return DatasourceReadResult( + datasource=descriptor, + rows=result.rows, + total_rows=result.total_rows, + truncated=result.truncated, + ) + + def _required_origin( + self, + session: Session, + principal: ApiPrincipal, + origin_ref: str, + ) -> DatasourceOrigin: + provider = datasource_origins(self._registry) + if provider is None: + raise DatasourceUnavailableError( + "No connector origin provider is currently available." + ) + try: + origin = provider.get_origin( + session, + principal, + origin_ref=origin_ref, + ) + except DatasourceError: + raise + except Exception as exc: + raise DatasourceUnavailableError( + f"The datasource origin could not be inspected: {exc}" + ) from exc + if origin is None: + raise DatasourceNotFoundError("Datasource origin not found.") + return origin + + def _read_origin_all( + self, + session: Session, + principal: ApiPrincipal, + *, + origin_ref: str, + ) -> tuple[tuple[Mapping[str, object], ...], DatasourceOrigin]: + provider = datasource_origins(self._registry) + if provider is None: + raise DatasourceUnavailableError( + "No connector origin provider is currently available." + ) + rows: list[Mapping[str, object]] = [] + origin: DatasourceOrigin | None = None + offset = 0 + while True: + try: + result = provider.read_origin( + session, + principal, + request=DatasourceOriginReadRequest( + origin_ref=origin_ref, + limit=min(MAX_READ_ROWS, MAX_STAGE_ROWS - offset), + offset=offset, + ), + ) + except DatasourceError: + raise + except Exception as exc: + raise DatasourceUnavailableError( + f"The datasource origin could not be materialized: {exc}" + ) from exc + origin = result.origin + rows.extend(result.rows) + offset += len(result.rows) + if not result.truncated: + break + if offset >= MAX_STAGE_ROWS or not result.rows: + raise DatasourceValidationError( + f"Cached and frozen sources are limited to {MAX_STAGE_ROWS:,} rows." + ) + if origin is None: + raise DatasourceUnavailableError("The datasource origin returned no metadata.") + return tuple(rows), origin + + def _materialize_origin( + self, + session: Session, + *, + item: DatasourceRecord, + origin: DatasourceOrigin, + rows: Sequence[Mapping[str, object]], + actor_id: str | None, + frozen: bool = False, + frozen_label: str | None = None, + set_current: bool, + ) -> DatasourceMaterializationRecord: + normalized = normalize_rows(rows) + schema = infer_schema(normalized) or origin.schema + fingerprint = fingerprint_rows(normalized, schema) + return _append_materialization( + session, + datasource=item, + rows=normalized, + schema=[field_payload(field) for field in schema], + fingerprint=fingerprint, + byte_count=encoded_size(normalized), + actor_id=actor_id, + frozen=frozen, + frozen_label=frozen_label, + source_timestamp=origin.updated_at, + provenance={ + **dict(item.provenance_), + "origin_ref": origin.ref, + "origin_fingerprint": origin.fingerprint, + "materialized_at": utcnow().isoformat(), + }, + metadata={ + **dict(item.metadata_), + "origin_schema_version": origin.schema_version, + }, + set_current=set_current, + ) + + +def _append_materialization( + session: Session, + *, + datasource: DatasourceRecord, + rows: Sequence[Mapping[str, object]], + schema: Sequence[Mapping[str, object]], + fingerprint: str, + byte_count: int, + actor_id: str | None, + frozen: bool = False, + frozen_label: str | None = None, + source_timestamp=None, + provenance: Mapping[str, object] | None = None, + metadata: Mapping[str, object] | None = None, + set_current: bool, +) -> DatasourceMaterializationRecord: + revision = ( + session.scalar( + select(func.max(DatasourceMaterializationRecord.revision)).where( + DatasourceMaterializationRecord.datasource_id == datasource.id + ) + ) + or 0 + ) + 1 + schema_payload = [dict(field) for field in schema] + schema_changed = datasource.schema_ != schema_payload + schema_version = ( + datasource.schema_version + 1 + if schema_changed + else datasource.schema_version + ) + materialization = DatasourceMaterializationRecord( + tenant_id=datasource.tenant_id, + datasource_id=datasource.id, + revision=revision, + state="published", + schema_version=max(1, int(schema_version or 1)), + schema_=schema_payload, + rows=[dict(row) for row in rows], + fingerprint=fingerprint, + row_count=len(rows), + byte_count=byte_count, + frozen_at=utcnow() if frozen else None, + frozen_label=_clean_optional(frozen_label), + source_timestamp=source_timestamp, + provenance_=dict(provenance or {}), + metadata_=dict(metadata or {}), + created_by=actor_id, + ) + session.add(materialization) + session.flush() + if set_current: + datasource.current_materialization_id = materialization.id + datasource.schema_ = schema_payload + datasource.schema_version = schema_version + datasource.fingerprint = fingerprint + datasource.row_count = materialization.row_count + datasource.byte_count = materialization.byte_count + datasource.updated_by = actor_id + session.flush() + return materialization + + +def _selected_materialization( + session: Session, + *, + item: DatasourceRecord, + request: DatasourceReadRequest, +) -> DatasourceMaterializationRecord | None: + if request.materialization_ref: + return _materialization_record( + session, + tenant_id=item.tenant_id, + datasource_id=item.id, + materialization_ref=request.materialization_ref, + ) + if request.consistency == "frozen": + return session.scalar( + select(DatasourceMaterializationRecord) + .where( + DatasourceMaterializationRecord.datasource_id == item.id, + DatasourceMaterializationRecord.tenant_id == item.tenant_id, + DatasourceMaterializationRecord.frozen_at.is_not(None), + ) + .order_by(DatasourceMaterializationRecord.revision.desc()) + .limit(1) + ) + return _current_materialization(session, item) + + +def _current_materialization( + session: Session, + item: DatasourceRecord, +) -> DatasourceMaterializationRecord | None: + if not item.current_materialization_id: + return None + return session.scalar( + select(DatasourceMaterializationRecord).where( + DatasourceMaterializationRecord.id == item.current_materialization_id, + DatasourceMaterializationRecord.datasource_id == item.id, + DatasourceMaterializationRecord.tenant_id == item.tenant_id, + ) + ) + + +def _datasource_record( + session: Session, + *, + tenant_id: str, + datasource_ref: str, +) -> DatasourceRecord | None: + datasource_id = _strip_ref(datasource_ref, "datasource:") + if datasource_id is None: + return None + return session.scalar( + select(DatasourceRecord).where( + DatasourceRecord.id == datasource_id, + DatasourceRecord.tenant_id == tenant_id, + DatasourceRecord.deleted_at.is_(None), + ) + ) + + +def _required_datasource( + session: Session, + *, + tenant_id: str, + datasource_ref: str, +) -> DatasourceRecord: + item = _datasource_record( + session, + tenant_id=tenant_id, + datasource_ref=datasource_ref, + ) + if item is None: + raise DatasourceNotFoundError("Datasource not found.") + return item + + +def _materialization_record( + session: Session, + *, + tenant_id: str, + datasource_id: str, + materialization_ref: str, +) -> DatasourceMaterializationRecord | None: + materialization_id = _strip_ref(materialization_ref, "materialization:") + if materialization_id is None: + return None + return session.scalar( + select(DatasourceMaterializationRecord).where( + DatasourceMaterializationRecord.id == materialization_id, + DatasourceMaterializationRecord.tenant_id == tenant_id, + DatasourceMaterializationRecord.datasource_id == datasource_id, + ) + ) + + +def _required_stage( + session: Session, + *, + tenant_id: str, + stage_ref: str, +) -> DatasourceStageRecord: + stage_id = _strip_ref(stage_ref, "stage:") + if stage_id is None: + raise DatasourceNotFoundError("Datasource stage not found.") + item = session.scalar( + select(DatasourceStageRecord).where( + DatasourceStageRecord.id == stage_id, + DatasourceStageRecord.tenant_id == tenant_id, + ) + ) + if item is None: + raise DatasourceNotFoundError("Datasource stage not found.") + return item + + +def _ensure_source_name_available( + session: Session, + *, + tenant_id: str, + source_name: str, +) -> None: + existing = session.scalar( + select(DatasourceRecord.id).where( + DatasourceRecord.tenant_id == tenant_id, + DatasourceRecord.source_name == source_name, + ) + ) + if existing is not None: + raise DatasourceValidationError( + f"A datasource with key {source_name!r} already exists." + ) + + +def _datasource_dto(item: DatasourceRecord) -> DatasourceDescriptor: + capabilities = ["read", "preview", "freeze"] + if item.mode == "cached" and item.provider_ref: + capabilities.append("refresh") + return DatasourceDescriptor( + ref=_datasource_ref(item.id), + source_name=item.source_name, + name=item.name, + description=item.description, + kind=cast(Any, item.kind), + mode=cast(Any, item.mode), + shape=cast(Any, item.shape), + status=item.status, + provider=item.provider, + provider_ref=item.provider_ref, + schema=_fields(item.schema_), + schema_version=str(item.schema_version), + fingerprint=item.fingerprint, + current_materialization_ref=( + _materialization_ref(item.current_materialization_id) + if item.current_materialization_id + else None + ), + row_count=item.row_count, + byte_count=item.byte_count, + updated_at=item.updated_at, + capabilities=tuple(capabilities), + provenance=dict(item.provenance_), + metadata=dict(item.metadata_), + ) + + +def _materialization_dto( + item: DatasourceMaterializationRecord, +) -> DatasourceMaterialization: + return DatasourceMaterialization( + ref=_materialization_ref(item.id), + datasource_ref=_datasource_ref(item.datasource_id), + revision=item.revision, + state=item.state, + fingerprint=item.fingerprint, + schema=_fields(item.schema_), + row_count=item.row_count, + byte_count=item.byte_count, + frozen_at=item.frozen_at, + frozen_label=item.frozen_label, + source_timestamp=item.source_timestamp, + created_at=item.created_at, + provenance=dict(item.provenance_), + metadata=dict(item.metadata_), + ) + + +def _stage_dto(item: DatasourceStageRecord) -> DatasourceStage: + return DatasourceStage( + ref=_stage_ref(item.id), + name=item.name, + source_name=item.source_name, + kind=cast(Any, item.kind), + mode=cast(Any, item.mode), + shape=cast(Any, item.shape), + state=item.state, + target_datasource_ref=( + _datasource_ref(item.target_datasource_id) + if item.target_datasource_id + else None + ), + fingerprint=item.fingerprint, + schema=_fields(item.schema_), + row_count=item.row_count, + byte_count=item.byte_count, + validation=dict(item.validation_), + created_at=item.created_at, + promoted_at=item.promoted_at, + promoted_materialization_ref=( + _materialization_ref(item.promoted_materialization_id) + if item.promoted_materialization_id + else None + ), + provenance=dict(item.provenance_), + metadata=dict(item.metadata_), + ) + + +def _fields(payload: Sequence[Mapping[str, object]]) -> tuple[DatasourceField, ...]: + return tuple( + DatasourceField( + name=str(item.get("name", "")), + data_type=str(item.get("data_type", "unknown")), + nullable=bool(item.get("nullable", True)), + ) + for item in payload + ) + + +def _context( + session: object, + principal: object, + required_scope: str | Sequence[str], +) -> tuple[Session, ApiPrincipal]: + if not isinstance(session, Session): + raise TypeError("Datasource providers require a SQLAlchemy session.") + if not isinstance(principal, ApiPrincipal): + raise DatasourceAccessError("A tenant API principal is required.") + required_scopes = ( + (required_scope,) if isinstance(required_scope, str) else tuple(required_scope) + ) + if not ( + any(has_scope(principal, scope) for scope in required_scopes) + or has_scope(principal, ADMIN_SCOPE) + ): + raise DatasourceAccessError( + f"Missing one of the required scopes: {', '.join(required_scopes)}" + ) + return session, principal + + +def _validate_columns( + schema: Sequence[Mapping[str, object]], + columns: tuple[str, ...], +) -> None: + if not columns: + return + known = {str(field.get("name", "")) for field in schema} + unknown = [column for column in columns if column not in known] + if unknown: + raise DatasourceValidationError( + f"Unknown datasource columns: {', '.join(unknown)}" + ) + + +def _select_columns( + row: Mapping[str, object], + columns: tuple[str, ...], +) -> Mapping[str, object]: + if not columns: + return dict(row) + return {column: row.get(column) for column in columns} + + +def _strip_ref(value: str, prefix: str) -> str | None: + cleaned = str(value or "").strip() + if not cleaned: + return None + if cleaned.startswith(prefix): + return cleaned[len(prefix) :] + return cleaned if ":" not in cleaned else None + + +def _datasource_ref(item_id: str) -> str: + return f"datasource:{item_id}" + + +def _materialization_ref(item_id: str) -> str: + return f"materialization:{item_id}" + + +def _stage_ref(item_id: str) -> str: + return f"stage:{item_id}" + + +def _actor_id(principal: ApiPrincipal) -> str | None: + return principal.account_id or principal.membership_id or principal.identity_id + + +def _clean_optional(value: object | None) -> str | None: + cleaned = str(value or "").strip() + return cleaned or None + + +def _int_schema_version(value: str) -> int: + try: + return max(1, int(value)) + except (TypeError, ValueError): + return 1 + + +def _escape_like(value: str) -> str: + return value.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") + + +__all__ = [ + "ADMIN_SCOPE", + "CATALOGUE_READ_SCOPE", + "SOURCE_WRITE_SCOPE", + "STAGE_WRITE_SCOPE", + "SqlDatasourceProvider", +] diff --git a/src/govoplan_datasources/backend/tabular.py b/src/govoplan_datasources/backend/tabular.py new file mode 100644 index 0000000..497366f --- /dev/null +++ b/src/govoplan_datasources/backend/tabular.py @@ -0,0 +1,198 @@ +from __future__ import annotations + +import csv +import hashlib +import io +import json +from collections.abc import Mapping, Sequence +from datetime import date, datetime +from decimal import Decimal +from typing import Any + +from govoplan_core.core.datasources import DatasourceField, DatasourceValidationError + + +MAX_STAGE_ROWS = 10_000 +MAX_STAGE_BYTES = 5_000_000 +MAX_READ_ROWS = 500 + + +def normalize_rows( + rows: Sequence[Mapping[str, object]], +) -> tuple[dict[str, Any], ...]: + if len(rows) > MAX_STAGE_ROWS: + raise DatasourceValidationError( + f"Staging is limited to {MAX_STAGE_ROWS:,} rows." + ) + normalized = tuple(_json_row(row) for row in rows) + if encoded_size(normalized) > MAX_STAGE_BYTES: + raise DatasourceValidationError( + f"Staging is limited to {MAX_STAGE_BYTES // 1_000_000} MB." + ) + return normalized + + +def parse_csv_rows( + csv_text: str, + *, + delimiter: str = ",", +) -> tuple[dict[str, Any], ...]: + if delimiter not in {",", ";", "\t", "|"}: + raise DatasourceValidationError("Unsupported CSV delimiter.") + if len(csv_text.encode("utf-8")) > MAX_STAGE_BYTES: + raise DatasourceValidationError( + f"Staging is limited to {MAX_STAGE_BYTES // 1_000_000} MB." + ) + try: + reader = csv.reader(io.StringIO(csv_text), delimiter=delimiter) + raw_header = next(reader, None) + if not raw_header: + raise DatasourceValidationError("CSV input needs a non-empty header row.") + header = [ + str(name or "").removeprefix("\ufeff").strip() + for name in raw_header + ] + if any(not name for name in header): + raise DatasourceValidationError("CSV input needs a non-empty header row.") + if len(set(header)) != len(header): + raise DatasourceValidationError("CSV headers must be unique.") + rows: list[dict[str, object]] = [] + for line_number, values in enumerate(reader, start=2): + if len(values) != len(header): + raise DatasourceValidationError( + f"CSV row {line_number} has {len(values)} values; expected {len(header)}." + ) + rows.append( + { + name: _csv_value(value) + for name, value in zip(header, values, strict=True) + } + ) + except csv.Error as exc: + raise DatasourceValidationError(f"CSV input is invalid: {exc}") from exc + return normalize_rows(rows) + + +def infer_schema( + rows: Sequence[Mapping[str, object]], +) -> tuple[DatasourceField, ...]: + names: list[str] = [] + for row in rows: + for name in row: + if name not in names: + names.append(name) + result: list[DatasourceField] = [] + 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( + DatasourceField( + name=name, + data_type=data_type, + nullable=len(concrete) != len(values), + ) + ) + return tuple(result) + + +def fingerprint_rows( + rows: Sequence[Mapping[str, object]], + schema: Sequence[DatasourceField], +) -> str: + payload = { + "schema": [field_payload(field) for field in schema], + "rows": [dict(row) for row in rows], + } + encoded = json.dumps(payload, sort_keys=True, separators=(",", ":"), default=str) + return hashlib.sha256(encoded.encode("utf-8")).hexdigest() + + +def encoded_size(rows: Sequence[Mapping[str, object]]) -> int: + return len( + json.dumps(rows, sort_keys=True, separators=(",", ":"), default=str).encode( + "utf-8" + ) + ) + + +def field_payload(field: DatasourceField) -> dict[str, object]: + return { + "name": field.name, + "data_type": field.data_type, + "nullable": field.nullable, + } + + +def _json_row(row: Mapping[str, object]) -> dict[str, Any]: + normalized = {str(key).strip(): value for key, value in row.items()} + if not normalized or any(not key for key in normalized): + raise DatasourceValidationError("Every row needs named columns.") + try: + encoded = json.dumps(normalized, default=_json_value) + except (TypeError, ValueError) as exc: + raise DatasourceValidationError( + f"Datasource values must be JSON compatible: {exc}" + ) from exc + return json.loads(encoded) + + +def _json_value(value: object) -> object: + if isinstance(value, (date, datetime, Decimal)): + return str(value) + raise TypeError(f"{type(value).__name__} is not JSON serializable") + + +def _type_name(value: object) -> str: + if isinstance(value, bool): + return "boolean" + if isinstance(value, int): + return "integer" + if isinstance(value, (float, Decimal)): + return "number" + if isinstance(value, str): + return "string" + if isinstance(value, list): + return "array" + if isinstance(value, dict): + return "object" + return type(value).__name__.lower() + + +def _csv_value(value: str) -> object: + cleaned = value.strip() + if not cleaned: + return None + lowered = cleaned.lower() + if lowered == "true": + return True + if lowered == "false": + return False + if ( + cleaned.isdigit() + and (cleaned == "0" or not cleaned.startswith("0")) + ) or ( + cleaned.startswith("-") + and cleaned[1:].isdigit() + and (cleaned[1:] == "0" or not cleaned[1:].startswith("0")) + ): + return int(cleaned) + try: + return float(cleaned) + except ValueError: + return cleaned + + +__all__ = [ + "MAX_READ_ROWS", + "MAX_STAGE_BYTES", + "MAX_STAGE_ROWS", + "encoded_size", + "field_payload", + "fingerprint_rows", + "infer_schema", + "normalize_rows", + "parse_csv_rows", +] diff --git a/src/govoplan_datasources/py.typed b/src/govoplan_datasources/py.typed new file mode 100644 index 0000000..8b13789 --- /dev/null +++ b/src/govoplan_datasources/py.typed @@ -0,0 +1 @@ + diff --git a/tests/test_lifecycle.py b/tests/test_lifecycle.py new file mode 100644 index 0000000..abba19b --- /dev/null +++ b/tests/test_lifecycle.py @@ -0,0 +1,351 @@ +from __future__ import annotations + +import unittest + +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker + +from govoplan_core.auth import ApiPrincipal +from govoplan_core.core.access import PrincipalRef +from govoplan_core.core.datasources import ( + CAPABILITY_DATASOURCE_ORIGINS, + DatasourceAccessError, + DatasourceField, + DatasourceOrigin, + DatasourceOriginReadRequest, + DatasourceOriginReadResult, + DatasourceReadRequest, + DatasourceStageInput, +) +from govoplan_core.db.base import Base, utcnow +from govoplan_datasources.backend.db.models import ( + DatasourceMaterializationRecord, + DatasourceRecord, + DatasourceStageRecord, +) +from govoplan_datasources.backend.service import ( + CATALOGUE_READ_SCOPE, + SOURCE_WRITE_SCOPE, + STAGE_WRITE_SCOPE, + SqlDatasourceProvider, +) + + +def principal( + tenant_id: str = "tenant-1", + *, + scopes: tuple[str, ...] = ( + CATALOGUE_READ_SCOPE, + SOURCE_WRITE_SCOPE, + STAGE_WRITE_SCOPE, + ), +) -> ApiPrincipal: + return ApiPrincipal( + principal=PrincipalRef( + account_id="account-1", + membership_id="membership-1", + tenant_id=tenant_id, + scopes=frozenset(scopes), + ), + account=object(), + user=object(), + ) + + +class FakeOriginProvider: + def __init__(self) -> None: + self.rows = [{"id": 1, "name": "Initial"}] + + def _origin(self) -> DatasourceOrigin: + return DatasourceOrigin( + ref="snapshot:origin-1", + source_name="connector_cases", + name="Connector cases", + kind="database", + shape="tabular", + supported_modes=("live", "cached"), + provider="connectors.test", + schema=( + DatasourceField(name="id", data_type="integer", nullable=False), + DatasourceField(name="name", data_type="string", nullable=False), + ), + schema_version="1", + fingerprint=f"version-{len(self.rows)}-{self.rows[-1]['name']}", + row_count=len(self.rows), + updated_at=utcnow(), + ) + + def list_origins( + self, + _session: object, + _principal: object, + *, + query: str = "", + limit: int = 100, + ): + origin = self._origin() + return (origin,) if query.lower() in origin.name.lower() else () + + def get_origin( + self, + _session: object, + _principal: object, + *, + origin_ref: str, + ): + return self._origin() if origin_ref == "snapshot:origin-1" else None + + def read_origin( + self, + _session: object, + _principal: object, + *, + request: DatasourceOriginReadRequest, + ) -> DatasourceOriginReadResult: + origin = self.get_origin(None, None, origin_ref=request.origin_ref) + if origin is None: + raise AssertionError("Unexpected origin") + rows = self.rows[request.offset : request.offset + request.limit] + return DatasourceOriginReadResult( + origin=origin, + rows=tuple(dict(row) for row in rows), + total_rows=len(self.rows), + truncated=request.offset + len(rows) < len(self.rows), + ) + + +class FakeRegistry: + def __init__(self, origin_provider: FakeOriginProvider) -> None: + self.origin_provider = origin_provider + + def has_capability(self, name: str) -> bool: + return name == CAPABILITY_DATASOURCE_ORIGINS + + def capability(self, name: str) -> object: + if not self.has_capability(name): + raise KeyError(name) + return self.origin_provider + + +class DatasourceLifecycleTests(unittest.TestCase): + def setUp(self) -> None: + self.engine = create_engine("sqlite:///:memory:") + Base.metadata.create_all( + self.engine, + tables=[ + DatasourceRecord.__table__, + DatasourceMaterializationRecord.__table__, + DatasourceStageRecord.__table__, + ], + ) + self.Session = sessionmaker(bind=self.engine) + self.session = self.Session() + self.origins = FakeOriginProvider() + self.provider = SqlDatasourceProvider( + registry=FakeRegistry(self.origins), + ) + + def tearDown(self) -> None: + self.session.close() + Base.metadata.drop_all( + self.engine, + tables=[ + DatasourceStageRecord.__table__, + DatasourceMaterializationRecord.__table__, + DatasourceRecord.__table__, + ], + ) + self.engine.dispose() + + def test_static_stage_promote_update_and_frozen_read(self) -> None: + first_stage = self.provider.create_stage( + self.session, + principal(), + stage=DatasourceStageInput( + name="Monthly cases", + source_name="monthly_cases", + kind="upload", + mode="static", + shape="tabular", + rows=({"id": 1, "status": "new"},), + ), + ) + datasource, first = self.provider.promote_stage( + self.session, + principal(), + stage_ref=first_stage.ref, + ) + frozen = self.provider.freeze_datasource( + self.session, + principal(), + datasource_ref=datasource.ref, + label="Import evidence", + ) + second_stage = self.provider.create_stage( + self.session, + principal(), + stage=DatasourceStageInput( + name="Monthly cases", + source_name="monthly_cases", + kind="upload", + mode="static", + shape="tabular", + target_datasource_ref=datasource.ref, + rows=({"id": 2, "status": "checked", "note": "schema changed"},), + ), + ) + updated, second = self.provider.promote_stage( + self.session, + principal(), + stage_ref=second_stage.ref, + ) + self.session.commit() + + current = self.provider.read_datasource( + self.session, + principal(), + request=DatasourceReadRequest(datasource_ref=datasource.ref), + ) + frozen_result = self.provider.read_datasource( + self.session, + principal(), + request=DatasourceReadRequest( + datasource_ref=datasource.ref, + consistency="frozen", + ), + ) + + self.assertEqual(1, first.revision) + self.assertEqual(2, frozen.revision) + self.assertEqual(3, second.revision) + self.assertEqual( + [{"id": 2, "status": "checked", "note": "schema changed"}], + list(current.rows), + ) + self.assertEqual([{"id": 1, "status": "new"}], list(frozen_result.rows)) + self.assertEqual(second.ref, updated.current_materialization_ref) + self.assertEqual("2", updated.schema_version) + self.assertEqual( + "2", + current.datasource.schema_version, + ) + self.assertEqual(frozen.ref, frozen_result.materialization.ref) + + def test_live_reads_origin_and_cached_refresh_is_explicit(self) -> None: + live = self.provider.register_origin( + self.session, + principal(), + origin_ref="snapshot:origin-1", + name="Live cases", + source_name="live_cases", + mode="live", + ) + cached = self.provider.register_origin( + self.session, + principal(), + origin_ref="snapshot:origin-1", + name="Cached cases", + source_name="cached_cases", + mode="cached", + ) + self.session.commit() + + self.origins.rows = [ + {"id": 1, "name": "Initial"}, + {"id": 2, "name": "New"}, + ] + live_read = self.provider.read_datasource( + self.session, + principal(), + request=DatasourceReadRequest(datasource_ref=live.ref), + ) + cached_before = self.provider.read_datasource( + self.session, + principal(), + request=DatasourceReadRequest(datasource_ref=cached.ref), + ) + cached_live = self.provider.read_datasource( + self.session, + principal(), + request=DatasourceReadRequest( + datasource_ref=cached.ref, + consistency="live", + ), + ) + refreshed, materialization = self.provider.refresh_datasource( + self.session, + principal(), + datasource_ref=cached.ref, + ) + cached_after = self.provider.read_datasource( + self.session, + principal(), + request=DatasourceReadRequest(datasource_ref=cached.ref), + ) + + self.assertEqual(2, live_read.total_rows) + self.assertEqual(1, cached_before.total_rows) + self.assertEqual(2, cached_live.total_rows) + self.assertEqual(2, cached_after.total_rows) + self.assertEqual(materialization.ref, refreshed.current_materialization_ref) + + def test_tenant_and_scope_isolation(self) -> None: + stage = self.provider.create_stage( + self.session, + principal(), + stage=DatasourceStageInput( + name="Private", + source_name="private", + kind="upload", + mode="static", + shape="tabular", + rows=({"id": 1},), + ), + ) + datasource, _ = self.provider.promote_stage( + self.session, + principal(), + stage_ref=stage.ref, + ) + self.session.commit() + + self.assertEqual( + (), + self.provider.list_datasources(self.session, principal("tenant-2")), + ) + self.assertIsNone( + self.provider.get_datasource( + self.session, + principal("tenant-2"), + datasource_ref=datasource.ref, + ) + ) + with self.assertRaises(DatasourceAccessError): + self.provider.list_datasources( + self.session, + principal(scopes=()), + ) + + def test_stage_writer_can_list_stages_without_catalogue_scope(self) -> None: + writer = principal(scopes=(STAGE_WRITE_SCOPE,)) + stage = self.provider.create_stage( + self.session, + writer, + stage=DatasourceStageInput( + name="Pending import", + source_name="pending_import", + kind="upload", + mode="static", + shape="tabular", + rows=({"id": 1},), + ), + ) + + self.assertEqual( + (stage.ref,), + tuple(item.ref for item in self.provider.list_stages(self.session, writer)), + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_manifest.py b/tests/test_manifest.py new file mode 100644 index 0000000..c140e37 --- /dev/null +++ b/tests/test_manifest.py @@ -0,0 +1,32 @@ +from __future__ import annotations + +import unittest + +from govoplan_core.core.datasources import ( + CAPABILITY_DATASOURCE_CATALOGUE, + CAPABILITY_DATASOURCE_LIFECYCLE, +) +from govoplan_datasources.backend.manifest import get_manifest + + +class DatasourceManifestTests(unittest.TestCase): + def test_manifest_exposes_provider_neutral_contracts(self) -> None: + manifest = get_manifest() + + self.assertEqual(manifest.id, "datasources") + self.assertIn(CAPABILITY_DATASOURCE_CATALOGUE, manifest.capability_factories) + self.assertIn(CAPABILITY_DATASOURCE_LIFECYCLE, manifest.capability_factories) + self.assertIn( + "datasources.materializations", + {item.name for item in manifest.provides_interfaces}, + ) + requirement = next( + item + for item in manifest.requires_interfaces + if item.name == "connectors.datasource_origins" + ) + self.assertTrue(requirement.optional) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_migrations.py b/tests/test_migrations.py new file mode 100644 index 0000000..592c348 --- /dev/null +++ b/tests/test_migrations.py @@ -0,0 +1,47 @@ +from __future__ import annotations + +import tempfile +import unittest +from pathlib import Path + +from alembic.runtime.migration import MigrationContext +from sqlalchemy import create_engine, inspect + +from govoplan_core.db.migrations import migrate_database +from govoplan_datasources.backend.manifest import get_manifest + + +class DatasourceMigrationTests(unittest.TestCase): + def test_baseline_creates_datasource_tables_and_head(self) -> None: + with tempfile.TemporaryDirectory(prefix="govoplan-datasources-migration-") as directory: + url = f"sqlite:///{Path(directory) / 'datasources.db'}" + migrate_database( + database_url=url, + enabled_modules=("datasources",), + manifest_factories=(get_manifest,), + ) + engine = create_engine(url) + try: + with engine.connect() as connection: + self.assertIn( + "f3a8d2c7b1e0", + set(MigrationContext.configure(connection).get_current_heads()), + ) + self.assertEqual( + { + "datasource_catalogue", + "datasource_materializations", + "datasource_stages", + }, + { + name + for name in inspect(connection).get_table_names() + if name.startswith("datasource_") + }, + ) + finally: + engine.dispose() + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_tabular.py b/tests/test_tabular.py new file mode 100644 index 0000000..7afee8c --- /dev/null +++ b/tests/test_tabular.py @@ -0,0 +1,31 @@ +from __future__ import annotations + +import unittest + +from govoplan_core.core.datasources import DatasourceValidationError +from govoplan_datasources.backend.tabular import ( + fingerprint_rows, + infer_schema, + parse_csv_rows, +) + + +class DatasourceTabularTests(unittest.TestCase): + def test_csv_schema_and_fingerprint_are_stable(self) -> None: + rows = parse_csv_rows("id;name\n1;Ada\n2;\n", delimiter=";") + schema = infer_schema(rows) + + self.assertEqual(["id", "name"], [field.name for field in schema]) + self.assertTrue(schema[1].nullable) + self.assertEqual( + fingerprint_rows(rows, schema), + fingerprint_rows(rows, schema), + ) + + def test_csv_requires_headers(self) -> None: + with self.assertRaises(DatasourceValidationError): + parse_csv_rows("", delimiter=",") + + +if __name__ == "__main__": + unittest.main() diff --git a/webui/package.json b/webui/package.json new file mode 100644 index 0000000..29d87d6 --- /dev/null +++ b/webui/package.json @@ -0,0 +1,32 @@ +{ + "name": "@govoplan/datasources-webui", + "version": "0.1.14", + "private": true, + "type": "module", + "main": "src/index.ts", + "module": "src/index.ts", + "types": "src/index.ts", + "exports": { + ".": { + "types": "./src/index.ts", + "import": "./src/index.ts" + }, + "./styles/datasources.css": "./src/styles/datasources.css" + }, + "scripts": { + "typecheck": "tsc --noEmit" + }, + "peerDependencies": { + "@govoplan/core-webui": "^0.1.14", + "lucide-react": "^1.23.0", + "react": "^19.0.0", + "react-dom": "^19.0.0", + "react-router-dom": "^7.1.1", + "typescript": "^5.7.2" + }, + "peerDependenciesMeta": { + "@govoplan/core-webui": { + "optional": true + } + } +} diff --git a/webui/src/api/datasources.ts b/webui/src/api/datasources.ts new file mode 100644 index 0000000..40fb90f --- /dev/null +++ b/webui/src/api/datasources.ts @@ -0,0 +1,231 @@ +import { + apiFetch, + type ApiSettings +} from "@govoplan/core-webui"; + +export type DatasourceMode = "live" | "cached" | "static"; +export type DatasourceShape = "tabular" | "document" | "binary" | "directory" | "stream"; + +export type DatasourceField = { + name: string; + data_type: string; + nullable: boolean; +}; + +export type Datasource = { + ref: string; + source_name: string; + name: string; + description?: string | null; + kind: string; + mode: DatasourceMode; + shape: DatasourceShape; + status: string; + provider?: string | null; + provider_ref?: string | null; + schema: DatasourceField[]; + schema_version: string; + fingerprint: string; + current_materialization_ref?: string | null; + row_count?: number | null; + byte_count?: number | null; + updated_at?: string | null; + capabilities: string[]; + provenance: Record; + metadata: Record; +}; + +export type DatasourceMaterialization = { + ref: string; + datasource_ref: string; + revision: number; + state: string; + fingerprint: string; + schema: DatasourceField[]; + row_count?: number | null; + byte_count?: number | null; + frozen_at?: string | null; + frozen_label?: string | null; + source_timestamp?: string | null; + created_at?: string | null; + provenance: Record; + metadata: Record; +}; + +export type DatasourceStage = { + ref: string; + name: string; + source_name: string; + kind: string; + mode: "static" | "cached"; + shape: DatasourceShape; + state: string; + target_datasource_ref?: string | null; + fingerprint: string; + schema: DatasourceField[]; + row_count?: number | null; + byte_count?: number | null; + validation: Record; + created_at?: string | null; + promoted_at?: string | null; + promoted_materialization_ref?: string | null; + provenance: Record; + metadata: Record; +}; + +export type DatasourceOrigin = { + ref: string; + source_name: string; + name: string; + description?: string | null; + kind: string; + shape: DatasourceShape; + supported_modes: DatasourceMode[]; + provider: string; + schema: DatasourceField[]; + schema_version: string; + fingerprint: string; + row_count?: number | null; + byte_count?: number | null; + updated_at?: string | null; + capabilities: string[]; + metadata: Record; +}; + +export type DatasourcePreview = { + datasource: Datasource; + rows: Record[]; + total_rows: number; + truncated: boolean; + materialization?: DatasourceMaterialization | null; +}; + +export async function listDatasources( + settings: ApiSettings, + query = "" +): Promise { + const suffix = query.trim() ? `?query=${encodeURIComponent(query.trim())}` : ""; + const response = await apiFetch<{ datasources: Datasource[] }>( + settings, + `/api/v1/datasources${suffix}` + ); + return response.datasources; +} + +export async function listDatasourceStages( + settings: ApiSettings +): Promise { + const response = await apiFetch<{ stages: DatasourceStage[] }>( + settings, + "/api/v1/datasources/stages" + ); + return response.stages; +} + +export async function listDatasourceOrigins( + settings: ApiSettings +): Promise<{ available: boolean; origins: DatasourceOrigin[] }> { + return apiFetch(settings, "/api/v1/datasources/origins"); +} + +export async function listDatasourceMaterializations( + settings: ApiSettings, + datasourceRef: string +): Promise { + const response = await apiFetch<{ materializations: DatasourceMaterialization[] }>( + settings, + `/api/v1/datasources/${refId(datasourceRef)}/materializations` + ); + return response.materializations; +} + +export function previewDatasource( + settings: ApiSettings, + datasourceRef: string, + consistency: "current" | "live" | "frozen" = "current" +): Promise { + return apiFetch( + settings, + `/api/v1/datasources/${refId(datasourceRef)}/preview?limit=100&consistency=${consistency}` + ); +} + +export function createDatasourceStage( + settings: ApiSettings, + payload: { + name: string; + source_name: string; + description?: string | null; + mode: "static" | "cached"; + target_datasource_ref?: string | null; + } & ( + { format: "json"; rows: Record[] } + | { format: "csv"; csv_text: string; delimiter: string } + ) +): Promise { + return apiFetch(settings, "/api/v1/datasources/stages", { + method: "POST", + body: JSON.stringify(payload) + }); +} + +export function promoteDatasourceStage( + settings: ApiSettings, + stageRef: string, + payload: { freeze?: boolean; frozen_label?: string | null } = {} +): Promise<{ datasource: Datasource; materialization: DatasourceMaterialization }> { + return apiFetch(settings, `/api/v1/datasources/stages/${refId(stageRef)}/promote`, { + method: "POST", + body: JSON.stringify(payload) + }); +} + +export function registerDatasourceOrigin( + settings: ApiSettings, + payload: { + origin_ref: string; + name: string; + source_name: string; + mode: "live" | "cached"; + description?: string | null; + } +): Promise { + return apiFetch(settings, "/api/v1/datasources/origins/register", { + method: "POST", + body: JSON.stringify(payload) + }); +} + +export function refreshDatasource( + settings: ApiSettings, + datasourceRef: string +): Promise<{ datasource: Datasource; materialization: DatasourceMaterialization }> { + return apiFetch(settings, `/api/v1/datasources/${refId(datasourceRef)}/refresh`, { + method: "POST" + }); +} + +export function freezeDatasource( + settings: ApiSettings, + datasourceRef: string, + label: string +): Promise { + return apiFetch(settings, `/api/v1/datasources/${refId(datasourceRef)}/freeze`, { + method: "POST", + body: JSON.stringify({ label: label.trim() || null }) + }); +} + +export function retireDatasource( + settings: ApiSettings, + datasourceRef: string +): Promise<{ retired: boolean; datasource_ref: string }> { + return apiFetch(settings, `/api/v1/datasources/${refId(datasourceRef)}`, { + method: "DELETE" + }); +} + +function refId(ref: string): string { + const separator = ref.indexOf(":"); + return encodeURIComponent(separator >= 0 ? ref.slice(separator + 1) : ref); +} diff --git a/webui/src/features/datasources/DatasourcesPage.tsx b/webui/src/features/datasources/DatasourcesPage.tsx new file mode 100644 index 0000000..83ccb98 --- /dev/null +++ b/webui/src/features/datasources/DatasourcesPage.tsx @@ -0,0 +1,1122 @@ +import { + useCallback, + useEffect, + useMemo, + useState +} from "react"; +import { + Archive, + Database, + DatabaseZap, + Download, + Eye, + Layers3, + Plus, + RefreshCw, + Snowflake, + Trash2, + Upload +} from "lucide-react"; +import { + Button, + ConfirmDialog, + Dialog, + DismissibleAlert, + FormField, + IconButton, + LoadingFrame, + SegmentedControl, + StatusBadge, + hasScope, + isApiError, + type ApiSettings, + type AuthInfo +} from "@govoplan/core-webui"; +import { + createDatasourceStage, + freezeDatasource, + listDatasourceMaterializations, + listDatasourceOrigins, + listDatasourceStages, + listDatasources, + previewDatasource, + promoteDatasourceStage, + refreshDatasource, + registerDatasourceOrigin, + retireDatasource, + type Datasource, + type DatasourceMaterialization, + type DatasourceOrigin, + type DatasourcePreview, + type DatasourceStage +} from "../../api/datasources"; + +type CatalogueView = "catalogue" | "staging" | "origins"; +type AddKind = "upload" | "origin"; +type UploadFormat = "json" | "csv"; + +export default function DatasourcesPage({ + settings, + auth +}: { + settings: ApiSettings; + auth: AuthInfo; +}) { + const [view, setView] = useState("catalogue"); + const [datasources, setDatasources] = useState([]); + const [stages, setStages] = useState([]); + const [origins, setOrigins] = useState([]); + const [originsAvailable, setOriginsAvailable] = useState(false); + const [selectedDatasourceRef, setSelectedDatasourceRef] = useState(""); + const [selectedStageRef, setSelectedStageRef] = useState(""); + const [selectedOriginRef, setSelectedOriginRef] = useState(""); + const [preview, setPreview] = useState(null); + const [materializations, setMaterializations] = useState([]); + const [search, setSearch] = useState(""); + const [loading, setLoading] = useState(true); + const [detailLoading, setDetailLoading] = useState(false); + const [working, setWorking] = useState(false); + const [error, setError] = useState(""); + const [success, setSuccess] = useState(""); + const [addOpen, setAddOpen] = useState(false); + const [freezeOpen, setFreezeOpen] = useState(false); + const [freezeLabel, setFreezeLabel] = useState(""); + const [retireOpen, setRetireOpen] = useState(false); + + const canManage = hasScope(auth, "datasources:source:write") + || hasScope(auth, "datasources:source:admin"); + const canStage = hasScope(auth, "datasources:stage:write") + || hasScope(auth, "datasources:source:admin"); + + const reload = useCallback(async (preferredDatasourceRef?: string) => { + setLoading(true); + setError(""); + try { + const [nextDatasources, nextStages, originCatalogue] = await Promise.all([ + listDatasources(settings), + listDatasourceStages(settings), + listDatasourceOrigins(settings) + ]); + setDatasources(nextDatasources); + setStages(nextStages); + setOrigins(originCatalogue.origins); + setOriginsAvailable(originCatalogue.available); + setSelectedDatasourceRef((current) => { + const preferred = preferredDatasourceRef || current; + return nextDatasources.some((item) => item.ref === preferred) + ? preferred + : nextDatasources[0]?.ref ?? ""; + }); + setSelectedStageRef((current) => nextStages.some((item) => item.ref === current) + ? current + : nextStages[0]?.ref ?? ""); + setSelectedOriginRef((current) => originCatalogue.origins.some((item) => item.ref === current) + ? current + : originCatalogue.origins[0]?.ref ?? ""); + } catch (loadError) { + setError(apiErrorMessage(loadError)); + } finally { + setLoading(false); + } + }, [settings]); + + useEffect(() => { + void reload(); + }, [reload]); + + useEffect(() => { + if (!selectedDatasourceRef) { + setPreview(null); + setMaterializations([]); + return; + } + let cancelled = false; + setDetailLoading(true); + Promise.allSettled([ + previewDatasource(settings, selectedDatasourceRef), + listDatasourceMaterializations(settings, selectedDatasourceRef) + ]).then(([previewResult, materializationResult]) => { + if (cancelled) return; + if (previewResult.status === "fulfilled") { + setPreview(previewResult.value); + } else { + setPreview(null); + setError(apiErrorMessage(previewResult.reason)); + } + setMaterializations( + materializationResult.status === "fulfilled" + ? materializationResult.value + : [] + ); + setDetailLoading(false); + }); + return () => { + cancelled = true; + }; + }, [selectedDatasourceRef, settings]); + + const selectedDatasource = datasources.find((item) => item.ref === selectedDatasourceRef) ?? null; + const selectedStage = stages.find((item) => item.ref === selectedStageRef) ?? null; + const selectedOrigin = origins.find((item) => item.ref === selectedOriginRef) ?? null; + + const visibleDatasources = useMemo( + () => filterItems(datasources, search, (item) => + `${item.name} ${item.source_name} ${item.description ?? ""} ${item.mode} ${item.kind}`), + [datasources, search] + ); + const visibleStages = useMemo( + () => filterItems(stages, search, (item) => + `${item.name} ${item.source_name} ${item.state} ${item.mode}`), + [stages, search] + ); + const visibleOrigins = useMemo( + () => filterItems(origins, search, (item) => + `${item.name} ${item.source_name} ${item.provider} ${item.kind}`), + [origins, search] + ); + + const promoteStage = async () => { + if (!selectedStage || selectedStage.state !== "ready") return; + setWorking(true); + setError(""); + try { + const result = await promoteDatasourceStage(settings, selectedStage.ref); + setSuccess(`Promoted ${result.datasource.name} as revision ${result.materialization.revision}.`); + setView("catalogue"); + await reload(result.datasource.ref); + } catch (operationError) { + setError(apiErrorMessage(operationError)); + } finally { + setWorking(false); + } + }; + + const refreshSelected = async () => { + if (!selectedDatasource) return; + setWorking(true); + setError(""); + try { + const result = await refreshDatasource(settings, selectedDatasource.ref); + setSuccess(`Refreshed ${result.datasource.name} as revision ${result.materialization.revision}.`); + await reload(result.datasource.ref); + } catch (operationError) { + setError(apiErrorMessage(operationError)); + } finally { + setWorking(false); + } + }; + + const freezeSelected = async () => { + if (!selectedDatasource) return; + setWorking(true); + setError(""); + try { + const materialization = await freezeDatasource( + settings, + selectedDatasource.ref, + freezeLabel + ); + setSuccess(`Created frozen revision ${materialization.revision}.`); + setFreezeOpen(false); + setFreezeLabel(""); + await reload(selectedDatasource.ref); + } catch (operationError) { + setError(apiErrorMessage(operationError)); + } finally { + setWorking(false); + } + }; + + const retireSelected = async () => { + if (!selectedDatasource) return; + setWorking(true); + setError(""); + try { + await retireDatasource(settings, selectedDatasource.ref); + setSuccess(`Retired ${selectedDatasource.name}.`); + setRetireOpen(false); + await reload(); + } catch (operationError) { + setError(apiErrorMessage(operationError)); + } finally { + setWorking(false); + } + }; + + return ( +
+
+ + +
+
+ + {view === "catalogue" ? + : view === "staging" ? + : } + + + {view === "catalogue" + ? selectedDatasource?.name ?? "Catalogue" + : view === "staging" + ? selectedStage?.name ?? "Staging" + : selectedOrigin?.name ?? "Connector origins"} + + + {view === "catalogue" + ? selectedDatasource?.source_name ?? "Governed data" + : view === "staging" + ? selectedStage?.source_name ?? "Inspect before promotion" + : selectedOrigin?.provider ?? "External acquisition"} + + + + + {view === "catalogue" && selectedDatasource?.mode === "cached" ? ( + + ) : null} + {view === "catalogue" && selectedDatasource ? ( + <> + + } + variant="danger" + onClick={() => setRetireOpen(true)} + disabled={!canManage || working} + /> + + ) : null} + {view === "staging" && selectedStage?.state === "ready" ? ( + + ) : null} + {view === "origins" && selectedOrigin ? ( + + ) : null} + +
+ +
+ {error ? ( + + {error} + + ) : null} + {success ? ( + + {success} + + ) : null} +
+ +
+ {view === "catalogue" ? ( + selectedDatasource ? ( + + ) : } label="No datasource selected" /> + ) : null} + {view === "staging" ? ( + selectedStage + ? + : } label="No stage selected" /> + ) : null} + {view === "origins" ? ( + selectedOrigin + ? + : } label={ + originsAvailable ? "No origin selected" : "Connectors unavailable" + } /> + ) : null} +
+ {working ?
Working...
: null} +
+
+ + setAddOpen(false)} + onCreated={async (result) => { + setAddOpen(false); + if ("state" in result) { + setView("staging"); + setSelectedStageRef(result.ref); + setSuccess(`Created stage ${result.name}. Review it before promotion.`); + await reload(); + } else { + setView("catalogue"); + setSuccess(`Registered ${result.name}.`); + await reload(result.ref); + } + }} + /> + { + if (!working) setFreezeOpen(false); + }} + footer={( + <> + + + + )} + > +

+ Create an immutable, addressable state for reproducible runs and evidence. +

+ + setFreezeLabel(event.target.value)} + placeholder="Optional evidence label" + /> + + + setRetireOpen(false)} + onConfirm={() => void retireSelected()} + /> +
+ ); +} + +function DatasourceDetail({ + datasource, + preview, + materializations, + loading +}: { + datasource: Datasource; + preview: DatasourcePreview | null; + materializations: DatasourceMaterialization[]; + loading: boolean; +}) { + return ( + <> +
+ + + + + +
+ {datasource.description ? ( +
{datasource.description}
+ ) : null} +
+
+ Preview + {preview ? `${formatNumber(preview.total_rows)} total rows` : ""} +
+ {loading ? ( +
Loading preview...
+ ) : preview ? ( + + ) : ( +
No preview available
+ )} +
+
+
+ Materializations + Immutable revisions +
+
+ + + + + + + + + + + + {materializations.map((item) => ( + + + + + + + + ))} + {!materializations.length ? ( + + ) : null} + +
RevisionCreatedRowsStateFingerprint
{item.revision}{formatDate(item.created_at)}{formatNumber(item.row_count)} + {item.frozen_at ? ( + + ) : } + + {shortFingerprint(item.fingerprint)} +
Live source without materializations
+
+
+
+
+ Schema + Version {datasource.schema_version} +
+ +
+ + ); +} + +function StageDetail({ stage }: { stage: DatasourceStage }) { + return ( + <> +
+ + + + + +
+
+
+ Validation + +
+
+ Fingerprint{shortFingerprint(stage.fingerprint)} + Target{stage.target_datasource_ref || "New datasource"} + Promoted revision{stage.promoted_materialization_ref || "Not promoted"} +
+
+
+
+ Detected schema + {stage.shape} +
+ +
+ + ); +} + +function OriginDetail({ origin }: { origin: DatasourceOrigin }) { + return ( + <> +
+ + + + + +
+ {origin.description ? ( +
{origin.description}
+ ) : null} +
+
+ Registration options + {origin.supported_modes.join(", ")} +
+
+ Origin reference{origin.ref} + Shape{origin.shape} + Fingerprint{shortFingerprint(origin.fingerprint)} +
+
+
+
+ Discovered schema + Version {origin.schema_version} +
+ +
+ + ); +} + +function AddDatasourceDialog({ + open, + settings, + initialKind, + initialOrigin, + origins, + originsAvailable, + datasources, + canStage, + canManage, + onClose, + onCreated +}: { + open: boolean; + settings: ApiSettings; + initialKind: AddKind; + initialOrigin: DatasourceOrigin | null; + origins: DatasourceOrigin[]; + originsAvailable: boolean; + datasources: Datasource[]; + canStage: boolean; + canManage: boolean; + onClose: () => void; + onCreated: (result: DatasourceStage | Datasource) => void | Promise; +}) { + const [kind, setKind] = useState(initialKind); + const [format, setFormat] = useState("csv"); + const [mode, setMode] = useState<"live" | "cached" | "static">( + initialKind === "origin" ? "live" : "static" + ); + const [originRef, setOriginRef] = useState(initialOrigin?.ref ?? ""); + const [targetRef, setTargetRef] = useState(""); + const [name, setName] = useState(initialOrigin?.name ?? ""); + const [sourceName, setSourceName] = useState(initialOrigin?.source_name ?? ""); + const [description, setDescription] = useState(initialOrigin?.description ?? ""); + const [rowsText, setRowsText] = useState('[\n { "id": 1 }\n]'); + const [csvText, setCsvText] = useState(""); + const [delimiter, setDelimiter] = useState(";"); + const [busy, setBusy] = useState(false); + const [error, setError] = useState(""); + + useEffect(() => { + if (!open) return; + setKind(initialKind); + setOriginRef(initialOrigin?.ref ?? ""); + setName(initialOrigin?.name ?? ""); + setSourceName(initialOrigin?.source_name ?? ""); + setDescription(initialOrigin?.description ?? ""); + setMode(initialKind === "origin" ? "live" : "static"); + setTargetRef(""); + setError(""); + }, [initialKind, initialOrigin, open]); + + const selectedOrigin = origins.find((item) => item.ref === originRef) ?? null; + + const chooseOrigin = (ref: string) => { + const origin = origins.find((item) => item.ref === ref); + setOriginRef(ref); + if (!origin) return; + setName(origin.name); + setSourceName(origin.source_name); + setDescription(origin.description ?? ""); + setMode(origin.supported_modes.includes("live") ? "live" : "cached"); + }; + + const chooseTarget = (ref: string) => { + setTargetRef(ref); + const target = datasources.find((item) => item.ref === ref); + if (!target) return; + setName(target.name); + setSourceName(target.source_name); + setDescription(target.description ?? ""); + setMode(target.mode === "cached" ? "cached" : "static"); + }; + + const create = async () => { + setBusy(true); + setError(""); + try { + if (kind === "origin") { + if (!selectedOrigin || (mode !== "live" && mode !== "cached")) { + throw new Error("Choose a connector origin and registration mode."); + } + const datasource = await registerDatasourceOrigin(settings, { + origin_ref: selectedOrigin.ref, + name, + source_name: sourceName, + description, + mode + }); + await onCreated(datasource); + } else { + const common = { + name, + source_name: sourceName, + description, + mode: mode === "cached" ? "cached" as const : "static" as const, + target_datasource_ref: targetRef || null + }; + const stage = format === "csv" + ? await createDatasourceStage(settings, { + ...common, + format: "csv", + csv_text: csvText, + delimiter + }) + : await createDatasourceStage(settings, { + ...common, + format: "json", + rows: parseRows(rowsText) + }); + await onCreated(stage); + } + } catch (createError) { + setError(apiErrorMessage(createError)); + } finally { + setBusy(false); + } + }; + + const loadFile = async (file: File | undefined) => { + if (!file) return; + try { + const text = await file.text(); + const isJson = file.name.toLowerCase().endsWith(".json"); + setFormat(isJson ? "json" : "csv"); + if (isJson) setRowsText(text); + else setCsvText(text); + const baseName = file.name.replace(/\.[^.]+$/, ""); + setName((current) => current || baseName); + setSourceName((current) => current || sourceNameFromFile(baseName)); + } catch { + setError("The selected file could not be read."); + } + }; + + return ( + { + if (!busy) onClose(); + }} + footer={( + <> + + + + )} + > +
+ {error ? ( + {error} + ) : null} + + ariaLabel="Datasource origin" + width="fill" + size="equal" + options={[ + { id: "upload", label: "Upload", disabled: !canStage }, + { + id: "origin", + label: "Connector origin", + disabled: !canManage || !originsAvailable + } + ]} + value={kind} + onChange={(next) => { + setKind(next); + setMode(next === "origin" ? "live" : "static"); + setError(""); + }} + /> + {kind === "origin" ? ( + + + + ) : ( + <> + + + + + void loadFile(event.target.files?.[0])} + /> + + + )} +
+ + setName(event.target.value)} /> + + + setSourceName(event.target.value)} + pattern="[A-Za-z_][A-Za-z0-9_]*" + placeholder="monthly_cases" + disabled={Boolean(targetRef)} + /> + +
+ + setDescription(event.target.value)} /> + + + setMode(next as typeof mode)} + /> + + {kind === "upload" ? ( + <> + + + ariaLabel="Upload format" + options={[ + { id: "json", label: "JSON" }, + { id: "csv", label: "CSV" } + ]} + value={format} + onChange={setFormat} + /> + + {format === "csv" ? ( + <> + + + + +