GovOPlaN/govoplan-datasources
1
0
Fork 0
Code Issues 5 Pull Requests Actions Packages Projects Releases Wiki Activity

feat: initialize governed datasources module

Browse Source
This commit is contained in:
Albrecht Degering
2026-07-28 12:43:02 +02:00
commit 1cb6228442
37 changed files with 5910 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
.gitea/ISSUE_TEMPLATE/bug_report.md Normal file
View File

@@ -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:

1
.gitea/ISSUE_TEMPLATE/config.yaml Normal file
View File

@@ -0,0 +1 @@
blank_issues_enabled: false

27
.gitea/ISSUE_TEMPLATE/docs_workflow.md Normal file
View File

@@ -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?

32
.gitea/ISSUE_TEMPLATE/feature_request.md Normal file
View File

@@ -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:

28
.gitea/ISSUE_TEMPLATE/task.md Normal file
View File

@@ -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:

25
.gitea/ISSUE_TEMPLATE/tech_debt.md Normal file
View File

@@ -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:

15
.gitea/PULL_REQUEST_TEMPLATE.md Normal file
View File

@@ -0,0 +1,15 @@
## Issue
Closes #
## Summary
-
## Verification
-
## Notes
Follow-up issues:

168
.gitignore vendored Normal file
View File

@@ -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/

25
README.md Normal file
View File

@@ -0,0 +1,25 @@
# govoplan-datasources
<!-- govoplan-repository-type:start -->
**Repository type:** module (platform).
<!-- govoplan-repository-type:end -->
`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.

63
docs/CONCEPT.md Normal file
View File

@@ -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.

22
pyproject.toml Normal file
View File

@@ -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"

3
src/govoplan_datasources/__init__.py Normal file
View File

@@ -0,0 +1,3 @@
"""GovOPlaN Datasources module."""
__version__ = "0.1.14"

1
src/govoplan_datasources/backend/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""Datasources backend."""

13
src/govoplan_datasources/backend/db/__init__.py Normal file
View File

@@ -0,0 +1,13 @@
"""Datasource database models."""
from govoplan_datasources.backend.db.models import (
DatasourceMaterializationRecord,
DatasourceRecord,
DatasourceStageRecord,
)
__all__ = [
"DatasourceMaterializationRecord",
"DatasourceRecord",
"DatasourceStageRecord",
]

217
src/govoplan_datasources/backend/db/models.py Normal file
View File

@@ -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",
]

287
src/govoplan_datasources/backend/manifest.py Normal file
View File

@@ -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",
]

1
src/govoplan_datasources/backend/migrations/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""Datasource Alembic migrations."""

1
src/govoplan_datasources/backend/migrations/versions/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""Datasource Alembic revisions."""

417
src/govoplan_datasources/backend/migrations/versions/f3a8d2c7b1e0_v0114_datasources_baseline.py Normal file
View File

@@ -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")

624
src/govoplan_datasources/backend/router.py Normal file
View File

@@ -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"]

16
src/govoplan_datasources/backend/runtime.py Normal file
View File

@@ -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"]

181
src/govoplan_datasources/backend/schemas.py Normal file
View File

@@ -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

1111
src/govoplan_datasources/backend/service.py Normal file
View File

File diff suppressed because it is too large Load Diff

198
src/govoplan_datasources/backend/tabular.py Normal file
View File

@@ -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",
]

1
src/govoplan_datasources/py.typed Normal file
View File

@@ -0,0 +1 @@

351
tests/test_lifecycle.py Normal file
View File

@@ -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()

32
tests/test_manifest.py Normal file
View File

@@ -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()

47
tests/test_migrations.py Normal file
View File

@@ -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()

31
tests/test_tabular.py Normal file
View File

@@ -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()

32
webui/package.json Normal file
View File

@@ -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
}
}
}

231
webui/src/api/datasources.ts Normal file
View File

@@ -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<string, unknown>;
metadata: Record<string, unknown>;
};
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<string, unknown>;
metadata: Record<string, unknown>;
};
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<string, unknown>;
created_at?: string | null;
promoted_at?: string | null;
promoted_materialization_ref?: string | null;
provenance: Record<string, unknown>;
metadata: Record<string, unknown>;
};
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<string, unknown>;
};
export type DatasourcePreview = {
datasource: Datasource;
rows: Record<string, unknown>[];
total_rows: number;
truncated: boolean;
materialization?: DatasourceMaterialization | null;
};
export async function listDatasources(
settings: ApiSettings,
query = ""
): Promise<Datasource[]> {
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<DatasourceStage[]> {
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<DatasourceMaterialization[]> {
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<DatasourcePreview> {
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<string, unknown>[] }
| { format: "csv"; csv_text: string; delimiter: string }
)
): Promise<DatasourceStage> {
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<Datasource> {
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<DatasourceMaterialization> {
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);
}

1122
webui/src/features/datasources/DatasourcesPage.tsx Normal file
View File

File diff suppressed because it is too large Load Diff

2
webui/src/index.ts Normal file
View File

@@ -0,0 +1,2 @@
export { datasourcesModule as default, datasourcesModule } from "./module";
export * from "./api/datasources";

40
webui/src/module.ts Normal file
View File

@@ -0,0 +1,40 @@
import { createElement, lazy } from "react";
import type { PlatformWebModule } from "@govoplan/core-webui";
import "./styles/datasources.css";
const DatasourcesPage = lazy(() => import("./features/datasources/DatasourcesPage"));
const readScopes = ["datasources:catalogue:read", "datasources:source:admin"];
export const datasourcesModule: PlatformWebModule = {
id: "datasources",
label: "Datasources",
version: "0.1.14",
optionalDependencies: [
"access",
"audit",
"connectors",
"files",
"notifications",
"policy"
],
navItems: [
{
to: "/datasources",
label: "Datasources",
iconName: "database-zap",
anyOf: readScopes,
order: 70
}
],
routes: [
{
path: "/datasources",
anyOf: readScopes,
order: 70,
render: ({ settings, auth }) => createElement(DatasourcesPage, { settings, auth })
}
]
};
export default datasourcesModule;

478
webui/src/styles/datasources.css Normal file
View File

@@ -0,0 +1,478 @@
.datasources-page {
position: relative;
height: calc(100vh - 115px);
min-width: 0;
min-height: 0;
padding: 0;
overflow: hidden;
color: var(--text);
background: var(--bg);
}
.datasources-page *,
.datasources-page *::before,
.datasources-page *::after {
box-sizing: border-box;
}
.datasources-shell {
display: grid;
grid-template-columns: minmax(250px, 300px) minmax(0, 1fr);
width: 100%;
height: 100%;
min-width: 0;
min-height: 0;
overflow: hidden;
border: var(--border-line);
background: var(--panel);
}
.datasources-sidebar,
.datasources-workspace,
.datasources-content,
.datasources-list-frame {
min-width: 0;
min-height: 0;
}
.datasources-sidebar {
display: flex;
flex-direction: column;
overflow: hidden;
border-right: var(--border-line);
background: var(--panel-soft);
}
.datasources-sidebar-toolbar,
.datasources-workspace-toolbar,
.datasources-section-heading {
display: flex;
align-items: center;
justify-content: space-between;
gap: 10px;
flex: 0 0 auto;
border-bottom: var(--border-line);
background: var(--panel-header);
}
.datasources-sidebar-toolbar {
min-height: 52px;
padding: 8px 10px 8px 14px;
}
.datasources-toolbar-actions,
.datasources-current-title,
.datasources-current-title > span,
.datasources-section-heading > span {
display: flex;
align-items: center;
gap: 7px;
}
.datasources-view-switch {
padding: 8px;
border-bottom: var(--border-line);
background: var(--panel);
}
.datasources-search {
padding: 9px;
border-bottom: var(--border-line);
}
.datasources-search input {
width: 100%;
min-height: 34px;
padding: 7px 9px;
}
.datasources-list-frame {
flex: 1 1 auto;
overflow: hidden;
}
.datasources-list {
height: 100%;
overflow: auto;
padding: 6px;
}
.datasources-list > button {
display: grid;
grid-template-columns: minmax(0, 1fr) auto;
align-items: center;
gap: 8px;
width: 100%;
min-height: 56px;
padding: 8px 9px;
border: 0;
border-radius: var(--radius-sm);
background: transparent;
color: var(--text);
cursor: pointer;
text-align: left;
}
.datasources-list > button:hover,
.datasources-list > button:focus-visible {
background: var(--primary-soft);
outline: none;
}
.datasources-list > button.is-selected {
background: var(--primary-soft-strong);
box-shadow: inset 3px 0 0 var(--accent);
}
.datasources-list > button > span:first-child {
min-width: 0;
}
.datasources-list strong,
.datasources-list small {
display: block;
overflow: hidden;
text-overflow: ellipsis;
white-space: nowrap;
}
.datasources-list strong {
color: var(--text-strong);
font-size: 13px;
}
.datasources-list small {
margin-top: 4px;
color: var(--muted);
font-size: 11px;
}
.datasources-list-empty,
.datasources-inline-empty,
.datasources-inline-loading {
display: grid;
place-items: center;
min-height: 110px;
color: var(--muted);
font-size: 13px;
text-align: center;
}
.datasources-workspace {
position: relative;
display: flex;
flex-direction: column;
overflow: hidden;
background: var(--bg);
}
.datasources-workspace-toolbar {
min-height: 58px;
padding: 8px 10px 8px 14px;
}
.datasources-current-title {
min-width: 0;
}
.datasources-current-title > span {
display: block;
min-width: 0;
}
.datasources-current-title strong,
.datasources-current-title small {
display: block;
overflow: hidden;
text-overflow: ellipsis;
white-space: nowrap;
}
.datasources-current-title strong {
color: var(--text-strong);
font-size: 14px;
}
.datasources-current-title small {
margin-top: 2px;
color: var(--muted);
font-size: 11px;
}
.datasources-alerts {
flex: 0 0 auto;
padding: 0 12px;
}
.datasources-alerts:empty {
display: none;
}
.datasources-alerts .alert {
margin: 10px 0 0;
}
.datasources-content {
flex: 1 1 auto;
overflow: auto;
padding: 14px;
}
.datasources-metrics {
display: grid;
grid-template-columns: repeat(5, minmax(100px, 1fr));
gap: 8px;
margin-bottom: 14px;
}
.datasources-metrics > span {
min-width: 0;
min-height: 61px;
padding: 9px 10px;
border: var(--border-line);
border-radius: var(--radius-sm);
background: var(--panel);
}
.datasources-metrics small,
.datasources-metrics strong {
display: block;
overflow: hidden;
text-overflow: ellipsis;
white-space: nowrap;
}
.datasources-metrics small {
color: var(--muted);
font-size: 10px;
text-transform: uppercase;
}
.datasources-metrics strong {
margin-top: 6px;
color: var(--text-strong);
font-size: 14px;
text-transform: capitalize;
}
.datasources-description {
margin-bottom: 14px;
padding: 10px 12px;
border-left: 3px solid var(--accent);
background: var(--panel-soft);
color: var(--text);
font-size: 13px;
line-height: 1.45;
}
.datasources-detail-section {
min-width: 0;
margin-bottom: 14px;
border: var(--border-line);
background: var(--panel);
}
.datasources-section-heading {
min-height: 42px;
padding: 7px 10px;
}
.datasources-section-heading > span {
color: var(--text-strong);
font-size: 13px;
font-weight: 650;
}
.datasources-section-heading small {
color: var(--muted);
font-size: 11px;
}
.datasources-table-scroll {
max-width: 100%;
overflow: auto;
}
.datasources-preview-table {
max-height: 310px;
}
.datasources-table-scroll table {
width: 100%;
border-collapse: collapse;
color: var(--text);
font-size: 12px;
}
.datasources-table-scroll th,
.datasources-table-scroll td {
min-width: 100px;
max-width: 320px;
padding: 8px 10px;
border-right: var(--border-line);
border-bottom: var(--border-line);
overflow: hidden;
text-align: left;
text-overflow: ellipsis;
white-space: nowrap;
}
.datasources-table-scroll th {
position: sticky;
top: 0;
z-index: 1;
background: var(--panel-header);
color: var(--text-strong);
font-weight: 650;
}
.datasources-table-scroll tr:last-child td {
border-bottom: 0;
}
.datasources-fingerprint {
font-family: ui-monospace, SFMono-Regular, Consolas, monospace;
}
.datasources-key-values {
display: grid;
grid-template-columns: repeat(3, minmax(0, 1fr));
gap: 0;
}
.datasources-key-values > span {
min-width: 0;
padding: 12px;
border-right: var(--border-line);
}
.datasources-key-values > span:last-child {
border-right: 0;
}
.datasources-key-values small,
.datasources-key-values strong {
display: block;
}
.datasources-key-values small {
margin-bottom: 5px;
color: var(--muted);
font-size: 10px;
text-transform: uppercase;
}
.datasources-key-values strong {
overflow-wrap: anywhere;
color: var(--text-strong);
font-size: 12px;
font-weight: 600;
}
.datasources-workspace-empty {
display: grid;
place-items: center;
align-content: center;
min-height: 100%;
gap: 10px;
color: var(--muted);
text-align: center;
}
.datasources-working {
position: absolute;
right: 12px;
bottom: 12px;
z-index: 4;
padding: 7px 10px;
border: var(--border-line);
border-radius: var(--radius-sm);
background: var(--panel);
box-shadow: var(--shadow);
color: var(--text-strong);
font-size: 12px;
}
.datasources-add-dialog {
width: min(760px, calc(100vw - 32px));
}
.datasources-dialog-fields {
display: grid;
gap: 12px;
}
.datasources-dialog-fields textarea {
width: 100%;
min-height: 180px;
resize: vertical;
font-family: ui-monospace, SFMono-Regular, Consolas, monospace;
font-size: 12px;
}
.datasources-dialog-grid {
display: grid;
grid-template-columns: minmax(0, 1fr) minmax(0, 1fr);
gap: 12px;
}
.datasources-dialog-copy {
margin: 0 0 14px;
color: var(--muted);
font-size: 13px;
line-height: 1.45;
}
@media (max-width: 900px) {
.datasources-page {
height: auto;
min-height: calc(100vh - 92px);
overflow: auto;
}
.datasources-shell {
grid-template-columns: 1fr;
height: auto;
overflow: visible;
}
.datasources-sidebar {
max-height: 42vh;
border-right: 0;
border-bottom: var(--border-line);
}
.datasources-workspace {
min-height: 58vh;
}
.datasources-metrics {
grid-template-columns: repeat(2, minmax(0, 1fr));
}
.datasources-workspace-toolbar {
align-items: flex-start;
flex-wrap: wrap;
}
.datasources-toolbar-actions {
flex-wrap: wrap;
}
}
@media (max-width: 560px) {
.datasources-metrics,
.datasources-key-values,
.datasources-dialog-grid {
grid-template-columns: 1fr;
}
.datasources-key-values > span {
border-right: 0;
border-bottom: var(--border-line);
}
.datasources-key-values > span:last-child {
border-bottom: 0;
}
}

1
webui/src/vite-env.d.ts vendored Normal file
View File

@@ -0,0 +1 @@
/// <reference path="../../../govoplan-core/webui/src/vite-env.d.ts" />

31
webui/tsconfig.json Normal file
View File

@@ -0,0 +1,31 @@
{
"compilerOptions": {
"target": "ES2020",
"useDefineForClassFields": true,
"lib": ["DOM", "DOM.Iterable", "ES2020"],
"allowJs": false,
"skipLibCheck": true,
"esModuleInterop": true,
"allowSyntheticDefaultImports": true,
"strict": true,
"noUnusedLocals": true,
"noUnusedParameters": true,
"forceConsistentCasingInFileNames": true,
"module": "ESNext",
"moduleResolution": "Bundler",
"resolveJsonModule": true,
"isolatedModules": true,
"noEmit": true,
"jsx": "react-jsx",
"preserveSymlinks": true,
"baseUrl": ".",
"paths": {
"@govoplan/core-webui": ["../../govoplan-core/webui/src/index.ts"],
"@govoplan/core-webui/*": ["../../govoplan-core/webui/src/*"],
"lucide-react": ["../../govoplan-core/webui/node_modules/lucide-react/dist/lucide-react.d.ts"],
"react": ["../../govoplan-core/webui/node_modules/@types/react/index.d.ts"],
"react/jsx-runtime": ["../../govoplan-core/webui/node_modules/@types/react/jsx-runtime.d.ts"]
}
},
"include": ["src"]
}