7 Commits
Author SHA1 Message Date
zemion 97ca670bfe feat(datasources): publish durable artifact outputs
Module Package Release / publish-packages (push) Successful in 12s
2026-08-21 17:36:19 +02:00
zemion f03497bdaf feat(datasources): add governed DSAR coverage 2026-08-21 03:19:42 +02:00
zemion 1bb03c0f91 refactor(webui): adopt semantic workspace actions 2026-08-19 18:47:45 +02:00
zemion 45cc3cdc7a feat: classify datasources product area 2026-08-18 21:32:50 +02:00
zemion 32a32c9906 Adopt shared WebUI structural primitives 2026-08-18 13:17:31 +02:00
zemion c16bb51aa3 Adopt shared WebUI layout primitives 2026-08-18 10:42:51 +02:00
zemion 492306f6a7 Add permission-aware datasource search source 2026-08-06 16:06:17 +02:00
17 changed files with 2638 additions and 421 deletions
+23 -4
View File
@@ -22,8 +22,11 @@ explicit frozen states, previews, retirement, and atomic producer publication.
Origin discovery retains each provider's source mode, structured health, and Origin discovery retains each provider's source mode, structured health, and
declared pushdown support. Live previews preserve the provider's effective row, declared pushdown support. Live previews preserve the provider's effective row,
serialized-byte, and elapsed-time limits and its redacted diagnostics. serialized-byte, and elapsed-time limits and its redacted diagnostics.
Producer modules can append a bounded tabular result or create a new static Producer modules can append a bounded inline tabular result or pin a larger
datasource through an idempotent capability. The publication ledger retains the durable artifact through the same idempotent capability. Artifact references
declare a backend, locator, SHA-256 checksum, schema, fingerprint, row and byte
counts; the installed provider verifies integrity and serves bounded reads.
The publication ledger retains the
producer run, output materialization, provenance, and replay identity. On producer run, output materialization, provenance, and replay identity. On
PostgreSQL, a transaction-scoped advisory lock serializes each tenant, producer, PostgreSQL, a transaction-scoped advisory lock serializes each tenant, producer,
and idempotency identity before any output side effect, so retries from multiple and idempotency identity before any output side effect, so retries from multiple
@@ -37,7 +40,23 @@ materialization provenance. The supported contract is documented in
The contracts already model database, HTTP/REST, directory, file, feed, The contracts already model database, HTTP/REST, directory, file, feed,
document, binary, directory, and stream sources so providers can be added document, binary, directory, and stream sources so providers can be added
without changing consumers. Larger durable artifact-backed publications remain without changing consumers. Storage modules contribute artifact backends
a later storage-provider slice. through the provider-neutral `datasources.artifactBackends` capability; the
Datasources module never imports their internals.
See [docs/CONCEPT.md](docs/CONCEPT.md) for ownership and lifecycle details. See [docs/CONCEPT.md](docs/CONCEPT.md) for ownership and lifecycle details.
## Data-subject requests
Datasources publishes `privacy.dsar.datasources` for exact catalogue,
governance-reference, materialization, payload, stage, and publication
references and for minimized operator attribution. It never exports connector
references, locators, credentials, arbitrary rows, schemas, validation
samples, metadata, provenance bodies, checkpoints, replay material, or hashes.
The module does not guess subject identity by scanning schema-dependent tabular
payloads; the authoritative source module locates and corrects those facts.
Unpromoted stages and unreferenced payloads can be deleted idempotently.
Published or referenced state, immutable materializations, governance evidence,
holds, and operator attribution require data-steward review. Dataflow and
Reporting derivatives must be refreshed after the source correction.
+7
View File
@@ -104,6 +104,13 @@ Consumers should be able to request the governance explanation and dependency
impact separately from row access. Seeing catalogue metadata must not imply impact separately from row access. Seeing catalogue metadata must not imply
permission to read protected data. permission to read protected data.
When Search is installed, Datasources contributes catalogue entries as a native
search source. Only the stable catalogue identity, display name, description,
mode, shape, lifecycle state, classification, publication state, and authority
mode are indexed. Every result is re-authorized against the current tenant and
catalogue-read permission. Rows, schemas, connector references, credentials,
arbitrary metadata, and provenance remain outside the derived search index.
## Next Providers ## Next Providers
Connector providers should cover: Connector providers should cover:
+20
View File
@@ -96,6 +96,26 @@ This makes concurrent retries from separate API or worker nodes converge on the
same publication and materialization rather than relying on a late uniqueness same publication and materialization rather than relying on a late uniqueness
failure after output rows have already been persisted. failure after output rows have already been persisted.
## Durable artifact publications
Outputs larger than the inline row and byte limits use an immutable artifact
reference. The reference pins its backend and locator together with SHA-256
checksum, schema, datasource fingerprint, row count, byte count, media type,
and optional resume checkpoint. Datasources persists that reference as the
materialization payload and asks the installed Core-contract artifact backend
to verify it before creating catalogue state. Reads remain bounded and are
re-authorized by Datasources before reaching the backend.
Schema rules are evaluated by Datasources. Content-level rules such as
uniqueness or range require producer evidence bound to the exact payload
checksum and current quality-policy hash, including all evaluated rule IDs.
Missing or explicitly deferred evidence produces a `review_required`
publication and an immutable, addressable materialization, but it does not
replace the Datasource's current state. Valid warnings produce
`published_with_warnings`; failed evidence blocks the publication without a
catalogue side effect. These terminal states are preserved for Dataflow and
Workflow handoffs instead of being collapsed into generic success.
Approval authority, approval expiry, and retention/deletion execution remain Approval authority, approval expiry, and retention/deletion execution remain
separate work under `govoplan-datasources#2`. Until those contracts are added, separate work under `govoplan-datasources#2`. Until those contracts are added,
no JSON flag is treated as an approval and no stage is deleted automatically. no JSON flag is treated as an approval and no stage is deleted automatically.
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project] [project]
name = "govoplan-datasources" name = "govoplan-datasources"
version = "0.1.18" version = "0.1.19"
description = "Governed datasource catalogue, staging, and materialization lifecycle for GovOPlaN." description = "Governed datasource catalogue, staging, and materialization lifecycle for GovOPlaN."
readme = "README.md" readme = "README.md"
requires-python = ">=3.12" requires-python = ">=3.12"
+1 -1
View File
@@ -1,3 +1,3 @@
"""GovOPlaN Datasources module.""" """GovOPlaN Datasources module."""
__version__ = "0.1.18" __version__ = "0.1.19"
@@ -0,0 +1,743 @@
from __future__ import annotations
from collections.abc import Sequence
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Any
from sqlalchemy import or_
from sqlalchemy.orm import Session
from govoplan_core.core.dsar import (
DsarErasureActionRef,
DsarExecutionResultRef,
DsarRecordRef,
DsarSubjectRef,
dsar_capability_name,
)
from govoplan_datasources.backend.db.models import (
DatasourceGovernanceReferenceRecord,
DatasourceMaterializationRecord,
DatasourcePayloadRecord,
DatasourcePublicationRecord,
DatasourceRecord,
DatasourceStageRecord,
)
DATASOURCES_DSAR_CAPABILITY = dsar_capability_name("datasources")
_MAX_RECORDS = 5_000
_CONFLICT = object()
_DIRECT_ALIASES = {
"datasource_id": ("datasources.datasource", "datasources.catalogue"),
"governance_reference_id": ("datasources.governance_reference",),
"materialization_id": ("datasources.materialization",),
"payload_id": ("datasources.payload",),
"stage_id": ("datasources.stage",),
"publication_id": ("datasources.publication",),
}
_RESOURCE_MODELS = {
"datasource": DatasourceRecord,
"datasource_governance_reference": DatasourceGovernanceReferenceRecord,
"datasource_materialization": DatasourceMaterializationRecord,
"datasource_payload": DatasourcePayloadRecord,
"datasource_stage": DatasourceStageRecord,
"datasource_publication": DatasourcePublicationRecord,
}
@dataclass(frozen=True, slots=True)
class _Selectors:
account_id: str | None
identity_id: str | None
membership_id: str | None
direct: dict[str, str]
@property
def actor_ids(self) -> tuple[str, ...]:
return tuple(
value
for value in (self.account_id, self.identity_id, self.membership_id)
if value
)
@dataclass(frozen=True, slots=True)
class _Match:
resource_type: str
row: Any
category: str
class DatasourcesDsarProvider:
provider_id = "datasources"
module_id = "datasources"
def search_subject(
self,
session: object,
*,
tenant_id: str,
subject: DsarSubjectRef,
) -> Sequence[DsarRecordRef]:
db = _session(session)
selectors = _selectors(subject)
if selectors is None or not (selectors.actor_ids or selectors.direct):
return ()
direct = _direct_matches(db, tenant_id=tenant_id, selectors=selectors)
if direct is None:
return ()
if direct:
if selectors.actor_ids and not any(
_correlates(match, selectors.actor_ids) for match in direct
):
return ()
matches = direct
else:
matches = _canonical_matches(
db,
tenant_id=tenant_id,
actor_ids=selectors.actor_ids,
)
records: list[DsarRecordRef] = []
seen: set[tuple[str, str]] = set()
for match in matches:
key = (match.resource_type, str(match.row.id))
if key in seen:
continue
if len(records) >= _MAX_RECORDS:
raise ValueError(
"Datasources DSAR result limit exceeded; narrow the selectors."
)
seen.add(key)
records.append(_record(match))
return tuple(records)
def plan_erasure(
self,
session: object,
*,
tenant_id: str,
subject: DsarSubjectRef,
records: Sequence[DsarRecordRef],
) -> Sequence[DsarErasureActionRef]:
del tenant_id
_session(session)
if _selectors(subject) is None:
raise ValueError("Datasources DSAR subject selectors conflict.")
actions: list[DsarErasureActionRef] = []
for record in records:
_validate_record(record)
if record.category in {
"unpromoted_datasource_stage",
"unreferenced_datasource_payload",
}:
kind = "delete"
executable = True
rationale = (
"Remove transient Datasources content that has not become "
"immutable or referenced lifecycle evidence."
)
elif record.category == "datasource_operator_attribution":
kind = "retain"
executable = False
rationale = record.retention_reason or (
"Institutional data operations remain attributable."
)
else:
kind = "manual_review"
executable = False
rationale = (
"A data steward must correct the authoritative source and "
"review immutable revisions, holds, consumers, and evidence."
)
actions.append(
DsarErasureActionRef(
action_id=(
f"datasources:{kind}:{record.resource_type}:"
f"{record.resource_id}"
),
provider_id=self.provider_id,
module_id=self.module_id,
kind=kind,
resource_type=record.resource_type,
resource_id=record.resource_id,
title=(
f"Delete {record.title}"
if executable
else f"Review {record.title}"
),
rationale=rationale,
executable=executable,
irreversible=executable,
metadata={"record_category": record.category},
)
)
return tuple(actions)
def execute_erasure(
self,
session: object,
*,
tenant_id: str,
subject: DsarSubjectRef,
actions: Sequence[DsarErasureActionRef],
request_id: str,
) -> Sequence[DsarExecutionResultRef]:
db = _session(session)
selectors = _selectors(subject)
if selectors is None:
raise ValueError("Datasources DSAR subject selectors conflict.")
results: list[DsarExecutionResultRef] = []
for action in actions:
_validate_action(action)
if not action.executable:
results.append(
DsarExecutionResultRef(
action_id=action.action_id,
status="blocked",
summary=(
"Use governed source correction and Datasources "
"retention review before changing published state."
),
evidence={"request_id": request_id},
)
)
continue
if action.kind != "delete" or action.resource_type not in {
"datasource_stage",
"datasource_payload",
}:
raise ValueError("Datasources DSAR executable action is unsupported.")
model = _RESOURCE_MODELS[action.resource_type]
row = (
db.query(model)
.filter(model.tenant_id == tenant_id, model.id == action.resource_id)
.with_for_update()
.one_or_none()
)
if row is None:
status = "unchanged"
summary = "Transient Datasources row was already absent."
else:
match = _Match(action.resource_type, row, "execution")
if not (
_directly_targets(selectors, match)
or _correlates(match, selectors.actor_ids)
):
raise ValueError(
"Datasources DSAR action is not corroborated by the subject."
)
_assert_deletable(db, resource_type=action.resource_type, row=row)
db.delete(row)
db.flush()
status = "executed"
summary = "Transient, unreferenced Datasources row removed."
results.append(
DsarExecutionResultRef(
action_id=action.action_id,
status=status,
summary=summary,
evidence={"request_id": request_id},
)
)
return tuple(results)
def _direct_matches(
session: Session,
*,
tenant_id: str,
selectors: _Selectors,
) -> list[_Match] | None:
matches: list[_Match] = []
for selector, value in selectors.direct.items():
if selector == "datasource_id":
datasource = _one(
session,
DatasourceRecord,
tenant_id=tenant_id,
field="id",
value=value.removeprefix("datasource:"),
)
if datasource is None:
return None
current = _datasource_package(
session,
tenant_id=tenant_id,
datasource=datasource,
)
else:
model, resource_type = {
"governance_reference_id": (
DatasourceGovernanceReferenceRecord,
"datasource_governance_reference",
),
"materialization_id": (
DatasourceMaterializationRecord,
"datasource_materialization",
),
"payload_id": (DatasourcePayloadRecord, "datasource_payload"),
"stage_id": (DatasourceStageRecord, "datasource_stage"),
"publication_id": (
DatasourcePublicationRecord,
"datasource_publication",
),
}[selector]
row = _one(
session,
model,
tenant_id=tenant_id,
field="id",
value=_strip_prefix(value),
)
if row is None:
return None
current = [
_Match(
resource_type, row, _direct_category(session, resource_type, row)
)
]
matches.extend(current)
if len(matches) > _MAX_RECORDS:
raise ValueError(
"Datasources DSAR result limit exceeded; narrow the selectors."
)
roots = {
root
for match in matches
for root in _root_datasource_ids(session, match)
if root
}
if len(roots) > 1:
return None
return matches
def _datasource_package(
session: Session,
*,
tenant_id: str,
datasource: DatasourceRecord,
) -> list[_Match]:
matches = [_Match("datasource", datasource, "datasource_configuration")]
specs = (
(
DatasourceGovernanceReferenceRecord,
"datasource_governance_reference",
),
(DatasourceMaterializationRecord, "datasource_materialization"),
(DatasourceStageRecord, "datasource_stage"),
(DatasourcePublicationRecord, "datasource_publication"),
)
materializations: list[DatasourceMaterializationRecord] = []
for model, resource_type in specs:
rows = (
session.query(model)
.filter(
model.tenant_id == tenant_id,
(
model.target_datasource_id == datasource.id
if model is DatasourceStageRecord
else model.datasource_id == datasource.id
),
)
.order_by(model.id)
.limit(_MAX_RECORDS + 1)
.all()
)
if model is DatasourceMaterializationRecord:
materializations = rows
matches.extend(
_Match(
resource_type,
row,
(
"datasource_related_stage"
if resource_type == "datasource_stage"
else _direct_category(session, resource_type, row)
),
)
for row in rows
)
payload_ids = {row.payload_id for row in materializations if row.payload_id}
if payload_ids:
payloads = (
session.query(DatasourcePayloadRecord)
.filter(
DatasourcePayloadRecord.tenant_id == tenant_id,
DatasourcePayloadRecord.id.in_(payload_ids),
)
.order_by(DatasourcePayloadRecord.id)
.limit(_MAX_RECORDS + 1)
.all()
)
matches.extend(
_Match(
"datasource_payload",
row,
_direct_category(session, "datasource_payload", row),
)
for row in payloads
)
return matches
def _canonical_matches(
session: Session,
*,
tenant_id: str,
actor_ids: tuple[str, ...],
) -> list[_Match]:
if not actor_ids:
return []
specs = (
(
DatasourceRecord,
or_(
DatasourceRecord.created_by.in_(actor_ids),
DatasourceRecord.updated_by.in_(actor_ids),
),
"datasource",
),
(
DatasourceMaterializationRecord,
DatasourceMaterializationRecord.created_by.in_(actor_ids),
"datasource_materialization",
),
(
DatasourcePayloadRecord,
DatasourcePayloadRecord.created_by.in_(actor_ids),
"datasource_payload",
),
(
DatasourceStageRecord,
DatasourceStageRecord.created_by.in_(actor_ids),
"datasource_stage",
),
(
DatasourcePublicationRecord,
DatasourcePublicationRecord.created_by.in_(actor_ids),
"datasource_publication",
),
)
matches: list[_Match] = []
for model, condition, resource_type in specs:
rows = (
session.query(model)
.filter(model.tenant_id == tenant_id, condition)
.order_by(model.id)
.limit(_MAX_RECORDS + 1)
.all()
)
matches.extend(
_Match(resource_type, row, "datasource_operator_attribution")
for row in rows
)
if len(matches) > _MAX_RECORDS:
raise ValueError(
"Datasources DSAR result limit exceeded; narrow the selectors."
)
return matches
def _one(
session: Session,
model: Any,
*,
tenant_id: str,
field: str,
value: str,
) -> Any | None:
return (
session.query(model)
.filter(model.tenant_id == tenant_id, getattr(model, field) == value)
.one_or_none()
)
def _direct_category(session: Session, resource_type: str, row: Any) -> str:
if resource_type == "datasource_stage" and row.promoted_at is None:
return "unpromoted_datasource_stage"
if resource_type == "datasource_payload":
referenced = (
session.query(DatasourceMaterializationRecord.id)
.filter(
DatasourceMaterializationRecord.tenant_id == row.tenant_id,
DatasourceMaterializationRecord.payload_id == row.id,
)
.limit(1)
.count()
)
if not referenced:
return "unreferenced_datasource_payload"
return {
"datasource": "datasource_configuration",
"datasource_governance_reference": "datasource_governance_configuration",
"datasource_materialization": "immutable_datasource_materialization",
"datasource_payload": "referenced_datasource_payload",
"datasource_stage": "promoted_datasource_stage",
"datasource_publication": "immutable_datasource_publication",
}[resource_type]
def _root_datasource_ids(session: Session, match: _Match) -> set[str]:
row = match.row
if match.resource_type == "datasource":
return {row.id}
if match.resource_type == "datasource_stage":
return {row.target_datasource_id} if row.target_datasource_id else set()
if match.resource_type == "datasource_payload":
return {
value
for (value,) in session.query(DatasourceMaterializationRecord.datasource_id)
.filter(
DatasourceMaterializationRecord.tenant_id == row.tenant_id,
DatasourceMaterializationRecord.payload_id == row.id,
)
.all()
}
return {row.datasource_id}
def _correlates(match: _Match, actor_ids: tuple[str, ...]) -> bool:
row = match.row
return any(
str(getattr(row, field, "") or "") in actor_ids
for field in ("created_by", "updated_by")
)
def _directly_targets(selectors: _Selectors, match: _Match) -> bool:
row = match.row
selector, field = {
"datasource": ("datasource_id", "id"),
"datasource_governance_reference": (
"governance_reference_id",
"id",
),
"datasource_materialization": ("materialization_id", "id"),
"datasource_payload": ("payload_id", "id"),
"datasource_stage": ("stage_id", "id"),
"datasource_publication": ("publication_id", "id"),
}[match.resource_type]
value = _strip_prefix(selectors.direct.get(selector, ""))
if value == str(getattr(row, field)):
return True
datasource_selector = _strip_prefix(selectors.direct.get("datasource_id", ""))
return bool(
datasource_selector
and datasource_selector in _root_datasource_ids_for_row(match)
)
def _root_datasource_ids_for_row(match: _Match) -> set[str]:
row = match.row
if match.resource_type == "datasource":
return {row.id}
if match.resource_type == "datasource_stage":
return {row.target_datasource_id} if row.target_datasource_id else set()
if match.resource_type == "datasource_payload":
return set()
return {row.datasource_id}
def _assert_deletable(session: Session, *, resource_type: str, row: Any) -> None:
if resource_type == "datasource_stage":
if row.promoted_at is not None or row.promoted_materialization_id is not None:
raise ValueError("Promoted Datasource stages require manual review.")
return
referenced = (
session.query(DatasourceMaterializationRecord.id)
.filter(
DatasourceMaterializationRecord.tenant_id == row.tenant_id,
DatasourceMaterializationRecord.payload_id == row.id,
)
.limit(1)
.count()
)
if referenced:
raise ValueError("Referenced Datasource payloads require manual review.")
def _record(match: _Match) -> DsarRecordRef:
row = match.row
immutable = match.category == "datasource_operator_attribution"
return DsarRecordRef(
provider_id="datasources",
module_id="datasources",
resource_type=match.resource_type,
resource_id=str(row.id),
category=match.category,
title=_title(match.resource_type),
data={
key: value
for key, value in _record_data(match.resource_type, row).items()
if value is not None
},
observed_at=_observed_at(row),
immutable_evidence=immutable,
retention_reason=(
"Institutional datasource creation, publication, and revision "
"activity remains attributable for governance and audit."
if immutable
else None
),
source_path="/datasources",
)
def _record_data(resource_type: str, row: Any) -> dict[str, object]:
if resource_type == "datasource":
return {
"kind": row.kind,
"mode": row.mode,
"shape": row.shape,
"status": row.status,
"schema_version": row.schema_version,
"row_count": row.row_count,
"byte_count": row.byte_count,
"authority_mode": row.authority_mode,
"classification": row.classification,
"publication_state": row.publication_state,
"deleted_at": _iso(row.deleted_at),
"created_at": _iso(row.created_at),
"updated_at": _iso(row.updated_at),
}
if resource_type == "datasource_governance_reference":
return {"relation": row.relation}
if resource_type == "datasource_materialization":
return {
"revision": row.revision,
"state": row.state,
"schema_version": row.schema_version,
"row_count": row.row_count,
"byte_count": row.byte_count,
"frozen_at": _iso(row.frozen_at),
"source_timestamp": _iso(row.source_timestamp),
"created_at": _iso(row.created_at),
}
if resource_type == "datasource_payload":
return {
"backend": row.backend,
"state": row.state,
"media_type": row.media_type,
"row_count": row.row_count,
"byte_count": row.byte_count,
"created_at": _iso(row.created_at),
}
if resource_type == "datasource_stage":
return {
"kind": row.kind,
"mode": row.mode,
"shape": row.shape,
"state": row.state,
"row_count": row.row_count,
"byte_count": row.byte_count,
"promoted": row.promoted_at is not None,
"promoted_at": _iso(row.promoted_at),
"created_at": _iso(row.created_at),
}
return {
"producer_module": row.producer_module,
"status": row.status,
"created_at": _iso(row.created_at),
}
def _selectors(subject: DsarSubjectRef) -> _Selectors | None:
references = subject.external_references
account_id = _coalesce(
subject.account_id,
references.get("datasources.account"),
references.get("access.account"),
)
identity_id = _coalesce(
subject.identity_id,
references.get("datasources.identity"),
references.get("identity.id"),
)
membership_id = _coalesce(
subject.membership_id,
references.get("datasources.membership"),
references.get("tenancy.membership"),
)
if any(value is _CONFLICT for value in (account_id, identity_id, membership_id)):
return None
direct: dict[str, str] = {}
for selector, aliases in _DIRECT_ALIASES.items():
value = _coalesce(*(references.get(alias) for alias in aliases))
if value is _CONFLICT:
return None
if value:
direct[selector] = str(value)
return _Selectors(
account_id=_optional(account_id),
identity_id=_optional(identity_id),
membership_id=_optional(membership_id),
direct=direct,
)
def _coalesce(*values: str | None) -> str | None | object:
normalized = {str(value).strip() for value in values if str(value or "").strip()}
if len(normalized) > 1:
return _CONFLICT
return next(iter(normalized), None)
def _optional(value: object) -> str | None:
return value if isinstance(value, str) and value else None
def _strip_prefix(value: str) -> str:
return value.partition(":")[2] if ":" in value else value
def _title(resource_type: str) -> str:
return resource_type.replace("_", " ").title()
def _observed_at(row: Any) -> datetime | None:
for field in ("promoted_at", "source_timestamp", "updated_at", "created_at"):
value = getattr(row, field, None)
if isinstance(value, datetime):
return _aware(value)
return None
def _iso(value: datetime | None) -> str | None:
aware = _aware(value)
return aware.isoformat() if aware else None
def _aware(value: datetime | None) -> datetime | None:
if value is None or value.tzinfo is not None:
return value
return value.replace(tzinfo=timezone.utc)
def _session(value: object) -> Session:
if not isinstance(value, Session):
raise TypeError("Datasources DSAR requires a SQLAlchemy Session.")
return value
def _validate_record(record: DsarRecordRef) -> None:
if record.provider_id != "datasources" or record.module_id != "datasources":
raise ValueError("Datasources DSAR cannot plan a foreign provider record.")
if record.resource_type not in _RESOURCE_MODELS or not record.resource_id:
raise ValueError("Datasources DSAR record identity is invalid.")
def _validate_action(action: DsarErasureActionRef) -> None:
if action.provider_id != "datasources" or action.module_id != "datasources":
raise ValueError("Datasources DSAR cannot execute a foreign provider action.")
if action.resource_type not in _RESOURCE_MODELS or not action.action_id.startswith(
"datasources:"
):
raise ValueError("Datasources DSAR action identity is invalid.")
__all__ = ["DATASOURCES_DSAR_CAPABILITY", "DatasourcesDsarProvider"]
+178 -16
View File
@@ -7,16 +7,19 @@ from govoplan_core.core.access import (
CAPABILITY_AUTH_PRINCIPAL_RESOLVER, CAPABILITY_AUTH_PRINCIPAL_RESOLVER,
) )
from govoplan_core.core.datasources import ( from govoplan_core.core.datasources import (
CAPABILITY_DATASOURCE_ARTIFACT_BACKENDS,
CAPABILITY_DATASOURCE_CATALOGUE, CAPABILITY_DATASOURCE_CATALOGUE,
CAPABILITY_DATASOURCE_LIFECYCLE, CAPABILITY_DATASOURCE_LIFECYCLE,
CAPABILITY_DATASOURCE_ORIGINS, CAPABILITY_DATASOURCE_ORIGINS,
CAPABILITY_DATASOURCE_PUBLICATION, CAPABILITY_DATASOURCE_PUBLICATION,
datasource_artifact_backend_provider,
) )
from govoplan_core.core.module_guards import ( from govoplan_core.core.module_guards import (
drop_table_retirement_provider, drop_table_retirement_provider,
persistent_table_uninstall_guard, persistent_table_uninstall_guard,
) )
from govoplan_core.core.modules import ( from govoplan_core.core.modules import (
CapabilityDocumentation,
DocumentationLink, DocumentationLink,
DocumentationTopic, DocumentationTopic,
FrontendModule, FrontendModule,
@@ -28,6 +31,7 @@ from govoplan_core.core.modules import (
ModuleManifest, ModuleManifest,
NavItem, NavItem,
PermissionDefinition, PermissionDefinition,
ProductAreaContribution,
RoleTemplate, RoleTemplate,
) )
from govoplan_core.core.provider_governance import ( from govoplan_core.core.provider_governance import (
@@ -35,9 +39,17 @@ from govoplan_core.core.provider_governance import (
ModuleArchitectureDocumentation, ModuleArchitectureDocumentation,
ModuleMaturityEvidence, ModuleMaturityEvidence,
) )
from govoplan_core.core.search import SearchSourceProviderRegistration
from govoplan_core.core.views import ViewSurface from govoplan_core.core.views import ViewSurface
from govoplan_core.db.base import Base from govoplan_core.db.base import Base
from govoplan_datasources.backend.db import models as datasource_models from govoplan_datasources.backend.db import models as datasource_models
from govoplan_datasources.backend.dsar_provider import (
DATASOURCES_DSAR_CAPABILITY,
DatasourcesDsarProvider,
)
from govoplan_datasources.backend.search_source import (
create_datasources_search_source,
)
from govoplan_datasources.backend.service import ( from govoplan_datasources.backend.service import (
ADMIN_SCOPE, ADMIN_SCOPE,
CATALOGUE_READ_SCOPE, CATALOGUE_READ_SCOPE,
@@ -45,12 +57,13 @@ from govoplan_datasources.backend.service import (
STAGE_WRITE_SCOPE, STAGE_WRITE_SCOPE,
SqlDatasourceProvider, SqlDatasourceProvider,
) )
from govoplan_datasources.backend.payloads import ExternalArtifactPayloadBackend
MODULE_ID = "datasources" MODULE_ID = "datasources"
MODULE_NAME = "Datasources" MODULE_NAME = "Datasources"
MODULE_VERSION = "0.1.18" MODULE_VERSION = "0.1.19"
DATASOURCE_INTERFACE_VERSION = "0.1.0" DATASOURCE_INTERFACE_VERSION = "0.2.0"
ARCHITECTURE = ModuleArchitectureDeclaration( ARCHITECTURE = ModuleArchitectureDeclaration(
layer="data_reporting_integration", layer="data_reporting_integration",
@@ -75,7 +88,7 @@ ARCHITECTURE = ModuleArchitectureDeclaration(
), ),
known_limits=( known_limits=(
"Governance references are stable provider-neutral refs; dedicated selectors depend on the owning optional modules.", "Governance references are stable provider-neutral refs; dedicated selectors depend on the owning optional modules.",
"Quality and freshness policies are stored and snapshotted but enforcement remains provider-specific.", "External artifact content validation depends on an installed payload backend and checksum-bound producer evidence.",
), ),
supported_authority_modes=( supported_authority_modes=(
"native_authoritative", "native_authoritative",
@@ -172,7 +185,24 @@ def _router(context: ModuleContext):
def _provider(context: ModuleContext) -> SqlDatasourceProvider: def _provider(context: ModuleContext) -> SqlDatasourceProvider:
return SqlDatasourceProvider(registry=context.registry) provider = datasource_artifact_backend_provider(context.registry)
backends = (
tuple(
ExternalArtifactPayloadBackend(backend)
for backend in provider.artifact_backends()
)
if provider is not None
else ()
)
return SqlDatasourceProvider(
registry=context.registry,
payload_backends=backends,
)
def _dsar_provider(context: ModuleContext) -> DatasourcesDsarProvider:
del context
return DatasourcesDsarProvider()
def _tenant_summary(session, tenant_id: str) -> dict[str, int]: def _tenant_summary(session, tenant_id: str) -> dict[str, int]:
@@ -219,11 +249,13 @@ manifest = ModuleManifest(
"files", "files",
"notifications", "notifications",
"policy", "policy",
"search",
), ),
optional_capabilities=( optional_capabilities=(
CAPABILITY_AUTH_PRINCIPAL_RESOLVER, CAPABILITY_AUTH_PRINCIPAL_RESOLVER,
CAPABILITY_AUTH_PERMISSION_EVALUATOR, CAPABILITY_AUTH_PERMISSION_EVALUATOR,
CAPABILITY_DATASOURCE_ORIGINS, CAPABILITY_DATASOURCE_ORIGINS,
CAPABILITY_DATASOURCE_ARTIFACT_BACKENDS,
), ),
provides_interfaces=( provides_interfaces=(
ModuleInterfaceProvider( ModuleInterfaceProvider(
@@ -246,6 +278,7 @@ manifest = ModuleManifest(
name="datasources.publication", name="datasources.publication",
version=DATASOURCE_INTERFACE_VERSION, version=DATASOURCE_INTERFACE_VERSION,
), ),
ModuleInterfaceProvider(name=DATASOURCES_DSAR_CAPABILITY, version="0.1.0"),
), ),
requires_interfaces=( requires_interfaces=(
ModuleInterfaceRequirement( ModuleInterfaceRequirement(
@@ -254,6 +287,12 @@ manifest = ModuleManifest(
version_max_exclusive="1.0.0", version_max_exclusive="1.0.0",
optional=True, optional=True,
), ),
ModuleInterfaceRequirement(
name="search.source",
version_min="1.0.0",
version_max_exclusive="2.0.0",
optional=True,
),
), ),
permissions=PERMISSIONS, permissions=PERMISSIONS,
role_templates=ROLE_TEMPLATES, role_templates=ROLE_TEMPLATES,
@@ -286,13 +325,63 @@ manifest = ModuleManifest(
order=70, order=70,
), ),
), ),
product_areas=(
ProductAreaContribution(
id="data-assurance",
module_id=MODULE_ID,
label="i18n:govoplan-core.product_area.data_assurance",
icon="database-zap",
description="i18n:govoplan-core.product_area.data_assurance_description",
surface_ids=(
"datasources.nav.datasources",
"datasources.route.datasources",
),
order=60,
),
),
view_surfaces=( view_surfaces=(
ViewSurface(id="datasources.page", module_id=MODULE_ID, kind="route", label="Datasources", order=70), ViewSurface(
ViewSurface(id="datasources.catalogue", module_id=MODULE_ID, kind="section", label="Datasource catalogue", order=10), id="datasources.page",
ViewSurface(id="datasources.staging", module_id=MODULE_ID, kind="section", label="Datasource staging", order=20), module_id=MODULE_ID,
ViewSurface(id="datasources.origins", module_id=MODULE_ID, kind="section", label="Datasource origins", order=30), kind="route",
ViewSurface(id="datasources.governance", module_id=MODULE_ID, kind="action", label="Datasource governance", order=40), label="Datasources",
ViewSurface(id="datasources.preview", module_id=MODULE_ID, kind="section", label="Datasource preview and materializations", order=50), order=70,
),
ViewSurface(
id="datasources.catalogue",
module_id=MODULE_ID,
kind="section",
label="Datasource catalogue",
order=10,
),
ViewSurface(
id="datasources.staging",
module_id=MODULE_ID,
kind="section",
label="Datasource staging",
order=20,
),
ViewSurface(
id="datasources.origins",
module_id=MODULE_ID,
kind="section",
label="Datasource origins",
order=30,
),
ViewSurface(
id="datasources.governance",
module_id=MODULE_ID,
kind="action",
label="Datasource governance",
order=40,
),
ViewSurface(
id="datasources.preview",
module_id=MODULE_ID,
kind="section",
label="Datasource preview and materializations",
order=50,
),
), ),
), ),
route_factory=_router, route_factory=_router,
@@ -300,8 +389,24 @@ manifest = ModuleManifest(
CAPABILITY_DATASOURCE_CATALOGUE: _provider, CAPABILITY_DATASOURCE_CATALOGUE: _provider,
CAPABILITY_DATASOURCE_LIFECYCLE: _provider, CAPABILITY_DATASOURCE_LIFECYCLE: _provider,
CAPABILITY_DATASOURCE_PUBLICATION: _provider, CAPABILITY_DATASOURCE_PUBLICATION: _provider,
DATASOURCES_DSAR_CAPABILITY: _dsar_provider,
},
capability_documentation={
DATASOURCES_DSAR_CAPABILITY: CapabilityDocumentation(
label="Datasources data-subject request provider",
summary="Finds governed datasource copies without exposing credentials or row payloads.",
contract_version="0.1.0",
documentation_types=("admin", "user"),
audience=("privacy_officer", "data_steward", "user"),
),
}, },
tenant_summary_providers=(_tenant_summary,), tenant_summary_providers=(_tenant_summary,),
search_sources=(
SearchSourceProviderRegistration(
id="datasources.catalogue",
factory=create_datasources_search_source,
),
),
migration_spec=MigrationSpec( migration_spec=MigrationSpec(
module_id=MODULE_ID, module_id=MODULE_ID,
metadata=Base.metadata, metadata=Base.metadata,
@@ -334,6 +439,29 @@ manifest = ModuleManifest(
), ),
architecture=ARCHITECTURE, architecture=ARCHITECTURE,
documentation=( documentation=(
DocumentationTopic(
id="datasources.data-subject-requests",
title="Datasources data-subject requests",
summary="Identify governed datasource copies while preserving credential, payload, and immutable-evidence boundaries.",
body=(
"Datasources matches exact tenant-scoped catalogue, governance-reference, materialization, payload, stage, and publication identifiers plus minimized account, identity, or membership operator attribution. DSAR results never copy connector/provider references, locators, credentials, arbitrary rows, schemas, validation samples, metadata, provenance bodies, checkpoints, idempotency material, or hashes. Arbitrary tabular payloads are not scanned for identifiers because that would be incomplete, schema-dependent, and liable to disclose unrelated people; the authoritative source module must locate and correct subject facts. "
"An unpromoted stage or unreferenced payload can be deleted idempotently. Published catalogue state, promoted stages, referenced payloads, immutable materializations, governance references, publications, holds, and operator attribution require data-steward review and source correction. Downstream Dataflow and Reporting outputs must be refreshed after correction."
),
layer="configured",
documentation_types=("admin", "user"),
audience=("user", "operator", "module_admin", "data_steward", "auditor"),
related_modules=("core", "connectors", "dataflow", "reporting"),
order=69,
metadata={
"seed": True,
"help_contexts": [
"datasources.data-subject-requests",
"datasources.catalogue",
"datasources.staging",
"datasources.preview",
],
},
),
DocumentationTopic( DocumentationTopic(
id="datasources.lifecycle", id="datasources.lifecycle",
title="Datasource lifecycle", title="Datasource lifecycle",
@@ -350,7 +478,11 @@ manifest = ModuleManifest(
"reports, controls, and decisions. It preserves origin source mode, " "reports, controls, and decisions. It preserves origin source mode, "
"structured health, declared pushdown, and effective row, byte, and time " "structured health, declared pushdown, and effective row, byte, and time "
"limits for live previews. Governance metadata visibility does not " "limits for live previews. Governance metadata visibility does not "
"grant access to protected rows." "grant access to protected rows. When Search is enabled, the module "
"indexes only bounded catalogue labels and governance-safe facets, "
"then rechecks the current catalogue permission before returning a "
"result. Rows, schemas, connector references, credentials, arbitrary "
"metadata, and provenance are never copied into the search index."
), ),
layer="available", layer="available",
documentation_types=("admin", "user"), documentation_types=("admin", "user"),
@@ -362,6 +494,7 @@ manifest = ModuleManifest(
"files", "files",
"reporting", "reporting",
"risk_compliance", "risk_compliance",
"search",
), ),
order=70, order=70,
metadata={ metadata={
@@ -383,7 +516,10 @@ manifest = ModuleManifest(
"Authority mode states whether GovOPlaN, an external system, a synchronized projection, an overlay, or a linked reference " "Authority mode states whether GovOPlaN, an external system, a synchronized projection, an overlay, or a linked reference "
"controls the data. The authoritative source, owner, steward, responsible organization/function, schema owner, privacy " "controls the data. The authoritative source, owner, steward, responsible organization/function, schema owner, privacy "
"profile, retention policy, transfer agreement, legal basis, holds, correction procedure, purposes, official keys, and " "profile, retention policy, transfer agreement, legal basis, holds, correction procedure, purposes, official keys, and "
"known limits provide discoverable institutional context. Freshness and quality policies are typed JSON contracts retained " "known limits provide discoverable institutional context. A retention-policy reference identifies the owning Policy rule; it "
"does not itself delete materializations, override legal holds, or prove that a scheduler is active. A transfer-agreement "
"reference records the governed agreement for external exchange; it does not grant connector credentials, recipient access, "
"or authority to export classified rows. Freshness and quality policies are typed JSON contracts retained "
"with materialization evidence. Datasources enforces the declared bounded tabular stage rules and schema policy; origin-specific " "with materialization evidence. Datasources enforces the declared bounded tabular stage rules and schema policy; origin-specific "
"or consumer-specific controls still remain with the provider or consuming control that declares support. Metadata " "or consumer-specific controls still remain with the provider or consuming control that declares support. Metadata "
"visibility never grants row access." "visibility never grants row access."
@@ -391,7 +527,14 @@ manifest = ModuleManifest(
layer="available", layer="available",
documentation_types=("admin", "user"), documentation_types=("admin", "user"),
audience=("operator", "module_admin", "data_steward", "product_owner"), audience=("operator", "module_admin", "data_steward", "product_owner"),
related_modules=("policy", "organizations", "idm", "dataflow", "reporting", "risk_compliance"), related_modules=(
"policy",
"organizations",
"idm",
"dataflow",
"reporting",
"risk_compliance",
),
order=71, order=71,
metadata={ metadata={
"seed": True, "seed": True,
@@ -403,6 +546,8 @@ manifest = ModuleManifest(
"datasources.field.publication-state", "datasources.field.publication-state",
"datasources.field.freshness-policy", "datasources.field.freshness-policy",
"datasources.field.quality-policy", "datasources.field.quality-policy",
"datasources.field.retention-policy",
"datasources.field.transfer-agreement",
], ],
}, },
), ),
@@ -416,7 +561,11 @@ manifest = ModuleManifest(
"promotion. Updates compare the detected schema with the current target and classify each change as compatible, warning, or " "promotion. Updates compare the detected schema with the current target and classify each change as compatible, warning, or "
"breaking. Diagnostics expose counts and bounded row numbers, never field values. The policy version and hash, diagnostics, and " "breaking. Diagnostics expose counts and bounded row numbers, never field values. The policy version and hash, diagnostics, and "
"schema diff are copied into immutable materialization provenance when promotion succeeds. Producer publication applies the same " "schema diff are copied into immutable materialization provenance when promotion succeeds. Producer publication applies the same "
"gate before any catalogue effect, retains validation evidence on the immutable output revision, serializes a publication identity across PostgreSQL worker nodes before replay lookup, and emits a transactional terminal event. Approval and retention execution " "gate before any catalogue effect, retains validation evidence on the immutable output revision, serializes a publication identity across PostgreSQL worker nodes before replay lookup, and emits a transactional terminal event. Large outputs may instead supply a "
"durable provider-neutral artifact reference with a pinned locator, SHA-256 checksum, declared schema, fingerprint, and size. "
"The configured payload backend verifies the artifact and provides bounded reads. Content-level rules require checksum- and "
"policy-bound producer evidence; absent evidence creates an immutable review-required materialization without changing the "
"Datasource's current state. Warning and review-required outcomes are preserved for Workflow handoffs. Approval and retention execution "
"are not inferred from arbitrary JSON flags and remain separate governed lifecycle work." "are not inferred from arbitrary JSON flags and remain separate governed lifecycle work."
), ),
layer="available", layer="available",
@@ -429,7 +578,13 @@ manifest = ModuleManifest(
kind="repository", kind="repository",
), ),
), ),
related_modules=("policy", "approvals", "audit", "dataflow", "workflow_engine"), related_modules=(
"policy",
"approvals",
"audit",
"dataflow",
"workflow_engine",
),
order=72, order=72,
metadata={ metadata={
"seed": True, "seed": True,
@@ -442,6 +597,7 @@ manifest = ModuleManifest(
"limitations": [ "limitations": [
"Referential rules currently use a bounded embedded value set rather than reading another protected Datasource.", "Referential rules currently use a bounded embedded value set rather than reading another protected Datasource.",
"Approval authority and automatic retention execution are not part of the current stage contract.", "Approval authority and automatic retention execution are not part of the current stage contract.",
"Artifact bytes remain owned by their payload backend; Datasources stores an immutable reference and integrity evidence.",
], ],
}, },
), ),
@@ -461,7 +617,13 @@ manifest = ModuleManifest(
layer="available", layer="available",
documentation_types=("admin", "user"), documentation_types=("admin", "user"),
audience=("operator", "module_admin", "power_user", "product_owner"), audience=("operator", "module_admin", "power_user", "product_owner"),
related_modules=("connectors", "dataflow", "workflow_engine", "reporting", "audit"), related_modules=(
"connectors",
"dataflow",
"workflow_engine",
"reporting",
"audit",
),
order=73, order=73,
metadata={ metadata={
"seed": True, "seed": True,
@@ -9,6 +9,9 @@ from sqlalchemy import delete, func, insert, select
from sqlalchemy.orm import Session from sqlalchemy.orm import Session
from govoplan_core.core.datasources import ( from govoplan_core.core.datasources import (
DatasourceArtifactBackend,
DatasourceArtifactReference,
DatasourceField,
DatasourceUnavailableError, DatasourceUnavailableError,
DatasourceValidationError, DatasourceValidationError,
) )
@@ -116,6 +119,91 @@ class DatabaseRowsPayloadBackend:
) )
class ExternalArtifactPayloadBackend:
"""Adapts a Core artifact backend to persisted Datasources payload rows."""
def __init__(self, backend: DatasourceArtifactBackend) -> None:
self._backend = backend
self.backend = backend.backend
def read_rows(
self,
session: Session,
payload: DatasourcePayloadRecord,
*,
offset: int,
limit: int,
) -> Sequence[Mapping[str, object]]:
return self._backend.read_rows(
session,
tenant_id=payload.tenant_id,
artifact=_artifact_from_payload(payload),
offset=offset,
limit=limit,
)
def verify(
self,
session: Session,
payload: DatasourcePayloadRecord,
) -> None:
self._backend.verify(
session,
tenant_id=payload.tenant_id,
artifact=_artifact_from_payload(payload),
)
def delete(
self,
session: Session,
payload: DatasourcePayloadRecord,
) -> None:
self._backend.delete(
session,
tenant_id=payload.tenant_id,
artifact=_artifact_from_payload(payload),
)
def _artifact_from_payload(
payload: DatasourcePayloadRecord,
) -> DatasourceArtifactReference:
metadata = dict(payload.metadata_)
raw_schema = metadata.get("artifact_schema")
schema = tuple(
DatasourceField(
name=str(item.get("name") or ""),
data_type=str(item.get("data_type") or "unknown"),
nullable=bool(item.get("nullable", True)),
)
for item in raw_schema
if isinstance(item, Mapping)
) if isinstance(raw_schema, Sequence) and not isinstance(
raw_schema, (str, bytes)
) else ()
return DatasourceArtifactReference(
backend=payload.backend,
locator=str(payload.locator or ""),
checksum=payload.checksum,
row_count=payload.row_count,
byte_count=payload.byte_count,
schema=schema,
fingerprint=str(metadata.get("publication_fingerprint") or ""),
media_type=payload.media_type,
checkpoint=dict(payload.checkpoint_),
metadata={
str(key): value
for key, value in metadata.items()
if key not in {"artifact_schema", "artifact_validation"}
},
validation=(
dict(metadata.get("artifact_validation"))
if isinstance(metadata.get("artifact_validation"), Mapping)
else {}
),
)
class PayloadBackendRegistry: class PayloadBackendRegistry:
def __init__( def __init__(
self, self,
@@ -363,6 +451,7 @@ __all__ = [
"DATABASE_ROWS_BACKEND", "DATABASE_ROWS_BACKEND",
"DatabaseRowsPayloadBackend", "DatabaseRowsPayloadBackend",
"DatasourcePayloadBackend", "DatasourcePayloadBackend",
"ExternalArtifactPayloadBackend",
"PayloadBackendRegistry", "PayloadBackendRegistry",
"create_database_rows_payload", "create_database_rows_payload",
"create_external_payload_reference", "create_external_payload_reference",
@@ -0,0 +1,179 @@
from __future__ import annotations
from collections.abc import Mapping, Sequence
from urllib.parse import quote
from sqlalchemy import func, select
from sqlalchemy.orm import Session
from govoplan_core.auth import ApiPrincipal
from govoplan_core.core.modules import ModuleContext
from govoplan_core.core.search import (
SearchAuthorizationRequest,
SearchBackfillPage,
SearchBackfillRequest,
SearchDocument,
SearchResourceType,
)
from govoplan_datasources.backend.db.models import DatasourceRecord
from govoplan_datasources.backend.service import ADMIN_SCOPE, CATALOGUE_READ_SCOPE
PROVIDER_ID = "datasources.catalogue"
RESOURCE_TYPE = "datasource"
class DatasourcesSearchSource:
def resource_types(self) -> Sequence[SearchResourceType]:
return (
SearchResourceType(
provider_id=PROVIDER_ID,
module_id="datasources",
resource_type=RESOURCE_TYPE,
label="Datasources",
requires_authorization_recheck=True,
),
)
def backfill(
self,
session: object,
*,
request: SearchBackfillRequest,
) -> SearchBackfillPage:
_assert_source(request.provider_id, request.resource_type)
db = _session(session)
statement = select(DatasourceRecord).where(
DatasourceRecord.tenant_id == request.tenant_id,
DatasourceRecord.deleted_at.is_(None),
)
if request.cursor:
statement = statement.where(DatasourceRecord.id > request.cursor)
rows = list(
db.scalars(
statement.order_by(DatasourceRecord.id).limit(request.limit + 1)
).all()
)
has_more = len(rows) > request.limit
selected = rows[: request.limit]
high_watermark = db.scalar(
select(func.max(DatasourceRecord.updated_at)).where(
DatasourceRecord.tenant_id == request.tenant_id,
DatasourceRecord.deleted_at.is_(None),
)
)
return SearchBackfillPage(
documents=tuple(_document(row) for row in selected),
next_cursor=selected[-1].id if has_more and selected else None,
complete=not has_more,
high_watermark=high_watermark.isoformat() if high_watermark else None,
)
def authorize(
self,
session: object,
principal: object,
*,
requests: Sequence[SearchAuthorizationRequest],
) -> Mapping[str, bool]:
decisions = {item.reference.key: False for item in requests}
if not isinstance(principal, ApiPrincipal) or not (
principal.has(CATALOGUE_READ_SCOPE) or principal.has(ADMIN_SCOPE)
):
return decisions
db = _session(session)
eligible = [
request
for request in requests
if request.reference.tenant_id == principal.tenant_id
and request.reference.module_id == "datasources"
and request.reference.resource_type == RESOURCE_TYPE
]
resource_ids = {request.reference.resource_id for request in eligible}
available_ids = (
set(
db.scalars(
select(DatasourceRecord.id).where(
DatasourceRecord.tenant_id == principal.tenant_id,
DatasourceRecord.id.in_(resource_ids),
DatasourceRecord.deleted_at.is_(None),
)
).all()
)
if resource_ids
else set()
)
for request in eligible:
decisions[request.reference.key] = (
request.reference.resource_id in available_ids
)
return decisions
def create_datasources_search_source(
_context: ModuleContext,
) -> DatasourcesSearchSource:
return DatasourcesSearchSource()
def _document(row: DatasourceRecord) -> SearchDocument:
datasource_ref = quote(f"datasource:{row.id}", safe="")
return SearchDocument(
tenant_id=row.tenant_id,
module_id="datasources",
provider_id=PROVIDER_ID,
resource_type=RESOURCE_TYPE,
resource_id=row.id,
title=row.name,
url=f"/datasources?datasource={datasource_ref}",
summary=(row.description or row.source_name)[:4000],
keywords=tuple(
value[:200]
for value in (
row.source_name,
row.kind,
row.mode,
row.shape,
row.status,
row.classification,
row.publication_state,
)
if value
),
visibility="restricted",
acl_tokens=(
f"scope:{CATALOGUE_READ_SCOPE}",
f"scope:{ADMIN_SCOPE}",
),
metadata={
"kind": row.kind,
"mode": row.mode,
"shape": row.shape,
"status": row.status,
"classification": row.classification,
"publication_state": row.publication_state,
"authority_mode": row.authority_mode,
},
source_revision=f"{row.schema_version}:{row.updated_at.isoformat()}",
source_updated_at=row.updated_at,
requires_authorization_recheck=True,
)
def _assert_source(provider_id: str, resource_type: str) -> None:
if provider_id != PROVIDER_ID or resource_type != RESOURCE_TYPE:
raise ValueError("Unsupported Datasources search source.")
def _session(value: object) -> Session:
if not isinstance(value, Session):
raise TypeError("Datasources search requires a SQLAlchemy session.")
return value
__all__ = [
"DatasourcesSearchSource",
"PROVIDER_ID",
"RESOURCE_TYPE",
"create_datasources_search_source",
]
+288 -21
View File
@@ -14,6 +14,7 @@ from govoplan_core.audit.logging import audit_event
from govoplan_core.auth import ApiPrincipal, has_scope from govoplan_core.auth import ApiPrincipal, has_scope
from govoplan_core.core.datasources import ( from govoplan_core.core.datasources import (
DatasourceAccessError, DatasourceAccessError,
DatasourceArtifactReference,
DatasourceDescriptor, DatasourceDescriptor,
DatasourceError, DatasourceError,
DatasourceField, DatasourceField,
@@ -25,6 +26,7 @@ from govoplan_core.core.datasources import (
DatasourceOriginReadRequest, DatasourceOriginReadRequest,
DatasourcePublicationRequest, DatasourcePublicationRequest,
DatasourcePublicationResult, DatasourcePublicationResult,
DatasourcePublicationStatus,
DatasourceReadRequest, DatasourceReadRequest,
DatasourceReadResult, DatasourceReadResult,
DatasourceStage, DatasourceStage,
@@ -46,6 +48,7 @@ from govoplan_datasources.backend.payloads import (
DatasourcePayloadBackend, DatasourcePayloadBackend,
PayloadBackendRegistry, PayloadBackendRegistry,
create_database_rows_payload, create_database_rows_payload,
create_external_payload_reference,
payload_for_materialization, payload_for_materialization,
validate_payload_size, validate_payload_size,
) )
@@ -72,9 +75,11 @@ class _PreparedPublication:
producer_module: str producer_module: str
producer_run_ref: str producer_run_ref: str
idempotency_key: str idempotency_key: str
rows: tuple[dict[str, Any], ...] rows: tuple[dict[str, Any], ...] | None
artifact: DatasourceArtifactReference | None
schema: tuple[DatasourceField, ...] schema: tuple[DatasourceField, ...]
fingerprint: str fingerprint: str
row_count: int
byte_count: int byte_count: int
request_hash: str request_hash: str
@@ -120,9 +125,8 @@ class SqlDatasourceProvider:
tenant_id=api_principal.tenant_id, tenant_id=api_principal.tenant_id,
request=request, request=request,
) )
validation = validate_stage( validation, publication_status = _validate_publication(
rows=prepared.rows, prepared=prepared,
schema=prepared.schema,
quality_policy=governance.quality_policy, quality_policy=governance.quality_policy,
baseline_schema=baseline_schema, baseline_schema=baseline_schema,
) )
@@ -138,6 +142,12 @@ class SqlDatasourceProvider:
"Published output failed governed quality or schema validation" "Published output failed governed quality or schema validation"
f"{suffix}." f"{suffix}."
) )
reusable_payload = self._publication_payload(
db,
tenant_id=api_principal.tenant_id,
actor_id=actor_id,
prepared=prepared,
)
datasource = _publication_target( datasource = _publication_target(
db, db,
tenant_id=api_principal.tenant_id, tenant_id=api_principal.tenant_id,
@@ -146,11 +156,14 @@ class SqlDatasourceProvider:
prepared=prepared, prepared=prepared,
target=target, target=target,
governance=governance, governance=governance,
publish_as_current=(
request.set_current and publication_status != "review_required"
),
) )
materialization = _append_materialization( materialization = _append_materialization(
db, db,
datasource=datasource, datasource=datasource,
rows=prepared.rows, rows=prepared.rows or (),
schema=[field_payload(field) for field in prepared.schema], schema=[field_payload(field) for field in prepared.schema],
fingerprint=prepared.fingerprint, fingerprint=prepared.fingerprint,
byte_count=prepared.byte_count, byte_count=prepared.byte_count,
@@ -163,7 +176,15 @@ class SqlDatasourceProvider:
"publication_validation": validation, "publication_validation": validation,
}, },
metadata=dict(request.metadata), metadata=dict(request.metadata),
set_current=request.set_current, set_current=(
request.set_current and publication_status != "review_required"
),
reusable_payload=reusable_payload,
state=(
"review_required"
if publication_status == "review_required"
else "published"
),
) )
publication = _create_publication_record( publication = _create_publication_record(
db, db,
@@ -173,6 +194,7 @@ class SqlDatasourceProvider:
materialization=materialization, materialization=materialization,
request=request, request=request,
prepared=prepared, prepared=prepared,
status=publication_status,
) )
audit_event( audit_event(
db, db,
@@ -180,7 +202,7 @@ class SqlDatasourceProvider:
user_id=getattr(api_principal.user, "id", None) user_id=getattr(api_principal.user, "id", None)
or api_principal.account_id, or api_principal.account_id,
api_key_id=api_principal.api_key_id, api_key_id=api_principal.api_key_id,
action="datasource.publication.published", action=f"datasource.publication.{publication_status}",
object_type="datasource_publication", object_type="datasource_publication",
object_id=publication.id, object_id=publication.id,
details={ details={
@@ -189,7 +211,12 @@ class SqlDatasourceProvider:
"datasource_ref": _datasource_ref(datasource.id), "datasource_ref": _datasource_ref(datasource.id),
"materialization_ref": _materialization_ref(materialization.id), "materialization_ref": _materialization_ref(materialization.id),
"fingerprint": prepared.fingerprint, "fingerprint": prepared.fingerprint,
"row_count": len(prepared.rows), "row_count": prepared.row_count,
"payload_backend": (
prepared.artifact.backend
if prepared.artifact is not None
else "database_rows"
),
"policy_hash": validation["policy_hash"], "policy_hash": validation["policy_hash"],
"schema_classification": validation["schema_change"][ "schema_classification": validation["schema_change"][
"classification" "classification"
@@ -198,12 +225,52 @@ class SqlDatasourceProvider:
) )
return DatasourcePublicationResult( return DatasourcePublicationResult(
ref=_publication_ref(publication.id), ref=_publication_ref(publication.id),
status=publication.status, status=cast(DatasourcePublicationStatus, publication.status),
datasource=_datasource_dto(datasource), datasource=_datasource_dto(datasource),
materialization=_materialization_dto(materialization), materialization=_materialization_dto(materialization),
replayed=False, replayed=False,
) )
def _publication_payload(
self,
session: Session,
*,
tenant_id: str,
actor_id: str | None,
prepared: _PreparedPublication,
) -> DatasourcePayloadRecord | None:
artifact = prepared.artifact
if artifact is None:
return None
backend = self._payload_backends.require(artifact.backend)
payload = create_external_payload_reference(
session,
tenant_id=tenant_id,
backend=artifact.backend,
locator=artifact.locator,
checksum=artifact.checksum,
row_count=artifact.row_count,
byte_count=artifact.byte_count,
actor_id=actor_id,
media_type=artifact.media_type,
checkpoint=artifact.checkpoint,
metadata={
**dict(artifact.metadata),
"publication_fingerprint": artifact.fingerprint,
"artifact_schema": [
field_payload(field) for field in artifact.schema
],
"artifact_validation": dict(artifact.validation),
},
)
try:
backend.verify(session, payload)
except Exception:
session.delete(payload)
session.flush()
raise
return payload
def list_datasources( def list_datasources(
self, self,
session: object, session: object,
@@ -1056,6 +1123,7 @@ def _append_materialization(
metadata: Mapping[str, object] | None = None, metadata: Mapping[str, object] | None = None,
set_current: bool, set_current: bool,
reusable_payload: DatasourcePayloadRecord | None = None, reusable_payload: DatasourcePayloadRecord | None = None,
state: str = "published",
) -> DatasourceMaterializationRecord: ) -> DatasourceMaterializationRecord:
datasource = _lock_datasource_for_materialization(session, datasource) datasource = _lock_datasource_for_materialization(session, datasource)
revision = _allocate_materialization_revision(session, datasource) revision = _allocate_materialization_revision(session, datasource)
@@ -1082,7 +1150,7 @@ def _append_materialization(
tenant_id=datasource.tenant_id, tenant_id=datasource.tenant_id,
datasource_id=datasource.id, datasource_id=datasource.id,
revision=revision, revision=revision,
state="published", state=state,
schema_version=max(1, int(schema_version or 1)), schema_version=max(1, int(schema_version or 1)),
schema_=schema_payload, schema_=schema_payload,
payload_id=payload.id, payload_id=payload.id,
@@ -1618,17 +1686,47 @@ def _prepare_publication(
producer_run_ref=producer_run_ref, producer_run_ref=producer_run_ref,
idempotency_key=idempotency_key, idempotency_key=idempotency_key,
) )
normalized = normalize_rows(request.rows) if (request.rows is None) == (request.artifact is None):
raise DatasourceValidationError(
"A publication requires exactly one inline row payload or durable "
"artifact reference."
)
artifact = request.artifact
if artifact is None:
normalized: tuple[dict[str, Any], ...] | None = normalize_rows(
request.rows or ()
)
schema = infer_schema(normalized) schema = infer_schema(normalized)
fingerprint = fingerprint_rows(normalized, schema) fingerprint = fingerprint_rows(normalized, schema)
row_count = len(normalized)
byte_count = encoded_size(normalized)
else:
normalized = None
schema = _validated_artifact_schema(artifact)
fingerprint = _validated_sha256(
artifact.fingerprint,
label="Artifact publication fingerprints",
)
_validated_sha256(
artifact.checksum,
label="Artifact publication checksums",
)
if artifact.row_count < 0 or artifact.byte_count < 0:
raise DatasourceValidationError(
"Artifact publication sizes cannot be negative."
)
row_count = artifact.row_count
byte_count = artifact.byte_count
return _PreparedPublication( return _PreparedPublication(
producer_module=producer_module, producer_module=producer_module,
producer_run_ref=producer_run_ref, producer_run_ref=producer_run_ref,
idempotency_key=idempotency_key, idempotency_key=idempotency_key,
rows=normalized, rows=normalized,
artifact=artifact,
schema=schema, schema=schema,
fingerprint=fingerprint, fingerprint=fingerprint,
byte_count=encoded_size(normalized), row_count=row_count,
byte_count=byte_count,
request_hash=_publication_request_hash( request_hash=_publication_request_hash(
request, request,
normalized=normalized, normalized=normalized,
@@ -1637,6 +1735,147 @@ def _prepare_publication(
) )
def _validated_artifact_schema(
artifact: DatasourceArtifactReference,
) -> tuple[DatasourceField, ...]:
schema = tuple(artifact.schema)
names = [field.name.strip() for field in schema]
if not schema or any(not name for name in names):
raise DatasourceValidationError(
"Artifact publications require a non-empty schema."
)
if len(names) != len(set(names)):
raise DatasourceValidationError(
"Artifact publication schema field names must be unique."
)
return schema
def _validated_sha256(value: str, *, label: str) -> str:
cleaned = value.strip().casefold()
try:
valid = len(cleaned) == 64 and int(cleaned, 16) >= 0
except ValueError:
valid = False
if not valid:
raise DatasourceValidationError(f"{label} must be SHA-256 values.")
return cleaned
def _validate_publication(
*,
prepared: _PreparedPublication,
quality_policy: Mapping[str, object],
baseline_schema: Sequence[DatasourceField] | None,
) -> tuple[dict[str, object], DatasourcePublicationStatus]:
if prepared.artifact is None:
validation = validate_stage(
rows=prepared.rows or (),
schema=prepared.schema,
quality_policy=quality_policy,
baseline_schema=baseline_schema,
)
status: DatasourcePublicationStatus = (
"published_with_warnings"
if validation.get("warnings")
else "published"
)
return validation, status
raw_rules = quality_policy.get("rules", [])
rules = (
[item for item in raw_rules if isinstance(item, Mapping)]
if isinstance(raw_rules, Sequence)
and not isinstance(raw_rules, (str, bytes))
else []
)
metadata_rule_types = {"required_fields", "field"}
metadata_policy = {
**dict(quality_policy),
"rules": [
dict(item)
for item in rules
if str(item.get("type") or "") in metadata_rule_types
],
}
validation = validate_stage(
rows=(),
schema=prepared.schema,
quality_policy=metadata_policy,
baseline_schema=baseline_schema,
)
if validation["valid"] is not True:
return validation, "review_required"
evidence = dict(prepared.artifact.validation)
content_rule_ids = {
str(item.get("id") or f"rule-{index + 1}")
for index, item in enumerate(rules)
if str(item.get("type") or "") not in metadata_rule_types
}
policy_probe = validate_stage(
rows=(),
schema=prepared.schema,
quality_policy=quality_policy,
baseline_schema=baseline_schema,
)
evidence_rule_ids = {
str(item)
for item in evidence.get("rules_evaluated", [])
if str(item).strip()
} if isinstance(evidence.get("rules_evaluated"), Sequence) and not isinstance(
evidence.get("rules_evaluated"), (str, bytes)
) else set()
evidence_verified = (
not content_rule_ids
or (
evidence.get("valid") is True
and str(evidence.get("policy_hash") or "")
== str(policy_probe["policy_hash"])
and str(evidence.get("payload_checksum") or "").casefold()
== prepared.artifact.checksum.casefold()
and content_rule_ids.issubset(evidence_rule_ids)
)
)
requested_status = str(evidence.get("status") or "verified")
if evidence.get("valid") is False or requested_status == "failed":
errors = evidence.get("errors")
validation["valid"] = False
validation["errors"] = (
list(errors)
if isinstance(errors, Sequence)
and not isinstance(errors, (str, bytes))
else [
{
"severity": "error",
"code": "quality.artifact_validation",
"message": "Artifact validation evidence reports failure.",
}
]
)
return validation, "review_required"
evidence_warnings = evidence.get("warnings")
if isinstance(evidence_warnings, Sequence) and not isinstance(
evidence_warnings, (str, bytes)
):
validation["warnings"] = [
*list(validation.get("warnings", [])),
*list(evidence_warnings),
]
validation["artifact_evidence"] = {
"verified": evidence_verified,
"rules_required": sorted(content_rule_ids),
"rules_evaluated": sorted(evidence_rule_ids),
"payload_checksum": prepared.artifact.checksum.casefold(),
}
if not evidence_verified or requested_status == "review_required":
return validation, "review_required"
if validation.get("warnings") or requested_status == "warning":
return validation, "published_with_warnings"
return validation, "published"
def _validate_publication_identity( def _validate_publication_identity(
*, *,
producer_module: str, producer_module: str,
@@ -1730,7 +1969,7 @@ def _existing_publication_result(
) )
return DatasourcePublicationResult( return DatasourcePublicationResult(
ref=_publication_ref(publication.id), ref=_publication_ref(publication.id),
status=publication.status, status=cast(DatasourcePublicationStatus, publication.status),
datasource=_datasource_dto(datasource), datasource=_datasource_dto(datasource),
materialization=_materialization_dto(materialization), materialization=_materialization_dto(materialization),
replayed=True, replayed=True,
@@ -1746,6 +1985,7 @@ def _publication_target(
prepared: _PreparedPublication, prepared: _PreparedPublication,
target: DatasourceRecord | None, target: DatasourceRecord | None,
governance: DatasourceGovernance, governance: DatasourceGovernance,
publish_as_current: bool,
) -> DatasourceRecord: ) -> DatasourceRecord:
if target is not None: if target is not None:
datasource = target datasource = target
@@ -1782,9 +2022,9 @@ def _publication_target(
provider_ref=prepared.producer_run_ref, provider_ref=prepared.producer_run_ref,
schema_version=1, schema_version=1,
schema_=[field_payload(field) for field in prepared.schema], schema_=[field_payload(field) for field in prepared.schema],
fingerprint=prepared.fingerprint, fingerprint=prepared.fingerprint if publish_as_current else "",
row_count=len(prepared.rows), row_count=prepared.row_count if publish_as_current else None,
byte_count=prepared.byte_count, byte_count=prepared.byte_count if publish_as_current else None,
provenance_={ provenance_={
**dict(request.provenance), **dict(request.provenance),
"producer_module": prepared.producer_module, "producer_module": prepared.producer_module,
@@ -1859,6 +2099,7 @@ def _create_publication_record(
materialization: DatasourceMaterializationRecord, materialization: DatasourceMaterializationRecord,
request: DatasourcePublicationRequest, request: DatasourcePublicationRequest,
prepared: _PreparedPublication, prepared: _PreparedPublication,
status: DatasourcePublicationStatus,
) -> DatasourcePublicationRecord: ) -> DatasourcePublicationRecord:
publication = DatasourcePublicationRecord( publication = DatasourcePublicationRecord(
tenant_id=tenant_id, tenant_id=tenant_id,
@@ -1868,12 +2109,19 @@ def _create_publication_record(
request_hash=prepared.request_hash, request_hash=prepared.request_hash,
datasource_id=datasource.id, datasource_id=datasource.id,
materialization_id=materialization.id, materialization_id=materialization.id,
status="published", status=status,
details_={ details_={
"fingerprint": prepared.fingerprint, "fingerprint": prepared.fingerprint,
"row_count": len(prepared.rows), "row_count": prepared.row_count,
"set_current": request.set_current, "set_current": (
request.set_current and status != "review_required"
),
"frozen": request.freeze, "frozen": request.freeze,
"payload_backend": (
prepared.artifact.backend
if prepared.artifact is not None
else "database_rows"
),
"validation": materialization.provenance_.get( "validation": materialization.provenance_.get(
"publication_validation", "publication_validation",
{}, {},
@@ -1889,7 +2137,7 @@ def _create_publication_record(
def _publication_request_hash( def _publication_request_hash(
request: DatasourcePublicationRequest, request: DatasourcePublicationRequest,
*, *,
normalized: Sequence[Mapping[str, object]], normalized: Sequence[Mapping[str, object]] | None,
fingerprint: str, fingerprint: str,
) -> str: ) -> str:
payload = { payload = {
@@ -1899,7 +2147,26 @@ def _publication_request_hash(
"name": request.name, "name": request.name,
"source_name": request.source_name, "source_name": request.source_name,
"description": request.description, "description": request.description,
"rows": [dict(row) for row in normalized], "rows": [dict(row) for row in (normalized or ())],
"artifact": (
{
"backend": request.artifact.backend,
"locator": request.artifact.locator,
"checksum": request.artifact.checksum.casefold(),
"row_count": request.artifact.row_count,
"byte_count": request.artifact.byte_count,
"schema": [
field_payload(field) for field in request.artifact.schema
],
"fingerprint": request.artifact.fingerprint.casefold(),
"media_type": request.artifact.media_type,
"checkpoint": dict(request.artifact.checkpoint),
"metadata": dict(request.artifact.metadata),
"validation": dict(request.artifact.validation),
}
if request.artifact is not None
else None
),
"fingerprint": fingerprint, "fingerprint": fingerprint,
"freeze": request.freeze, "freeze": request.freeze,
"frozen_label": request.frozen_label, "frozen_label": request.frozen_label,
+581
View File
@@ -0,0 +1,581 @@
from __future__ import annotations
import json
import unittest
from datetime import UTC, datetime
from sqlalchemy import create_engine
from sqlalchemy.orm import Session
from govoplan_core.core.dsar import (
DsarErasureActionRef,
DsarProvider,
DsarRecordRef,
DsarSubjectRef,
)
from govoplan_core.db.base import Base
from govoplan_core.privacy.dsar_workflow import (
create_data_subject_request,
search_data_subject_request,
)
from govoplan_datasources.backend.db.models import (
DatasourceGovernanceReferenceRecord,
DatasourceMaterializationRecord,
DatasourcePayloadRecord,
DatasourcePayloadRowRecord,
DatasourcePublicationRecord,
DatasourceRecord,
DatasourceStageRecord,
)
from govoplan_datasources.backend.dsar_provider import (
DATASOURCES_DSAR_CAPABILITY,
DatasourcesDsarProvider,
)
from govoplan_datasources.backend.manifest import manifest
NOW = datetime(2026, 8, 21, 19, 0, tzinfo=UTC)
SECRET = "private-datasource-detail-do-not-export"
class _Registry:
def __init__(
self,
provider: DatasourcesDsarProvider,
*,
active: bool = True,
) -> None:
self.provider = provider
self.active = active
def capability_names(self):
return (DATASOURCES_DSAR_CAPABILITY,)
def capability_owner(self, name):
self._assert_capability(name)
return "datasources"
def tenant_entitlement_resolver(self):
active = self.active
class _Resolver:
@staticmethod
def resolve(session, tenant_id):
del session, tenant_id
return type(
"State",
(),
{"effective_modules": ("datasources",) if active else ()},
)()
return _Resolver()
def require_tenant_capability(self, name, session, **kwargs):
del session, kwargs
self._assert_capability(name)
return self.provider
def manifests(self):
return (type("Manifest", (), {"id": "datasources"})(),)
@staticmethod
def _assert_capability(name: str) -> None:
if name != DATASOURCES_DSAR_CAPABILITY:
raise KeyError(name)
class DatasourcesDsarProviderTests(unittest.TestCase):
def setUp(self) -> None:
self.engine = create_engine("sqlite+pysqlite:///:memory:")
Base.metadata.create_all(self.engine)
self.session = Session(self.engine)
self.provider = DatasourcesDsarProvider()
self.assertIsInstance(self.provider, DsarProvider)
self._seed()
self.session.commit()
def tearDown(self) -> None:
self.session.close()
self.engine.dispose()
def _seed(self) -> None:
datasource = self._datasource("datasource-1", "tenant-1")
other = self._datasource("datasource-other", "tenant-2")
self.session.add_all((datasource, other))
self.session.flush()
referenced_payload = self._payload(
"payload-referenced",
"tenant-1",
created_by="account-1",
)
orphan_payload = self._payload(
"payload-orphan",
"tenant-1",
created_by="account-1",
)
self.session.add_all((referenced_payload, orphan_payload))
self.session.flush()
self.session.add(
DatasourcePayloadRowRecord(
payload_id=referenced_payload.id,
row_index=0,
row_={"secret": SECRET},
checksum="a" * 64,
)
)
materialization = DatasourceMaterializationRecord(
id="materialization-1",
tenant_id="tenant-1",
datasource_id=datasource.id,
revision=1,
state="published",
schema_version=1,
schema_=[{"secret": SECRET}],
payload_id=referenced_payload.id,
payload_checksum="b" * 64,
rows=[{"secret": SECRET}],
fingerprint="c" * 64,
row_count=1,
byte_count=100,
source_timestamp=NOW,
provenance_={"secret": SECRET},
metadata_={"secret": SECRET},
governance_snapshot_={"secret": SECRET},
created_by="account-1",
)
self.session.add(materialization)
self.session.flush()
datasource.current_materialization_id = materialization.id
self.session.add_all(
(
DatasourceGovernanceReferenceRecord(
id="governance-1",
tenant_id="tenant-1",
datasource_id=datasource.id,
relation="authoritative_source",
reference=SECRET,
),
self._stage(
"stage-1",
target_datasource_id=datasource.id,
promoted=False,
),
self._stage(
"stage-promoted",
target_datasource_id=datasource.id,
promoted=True,
),
DatasourcePublicationRecord(
id="publication-1",
tenant_id="tenant-1",
producer_module="dataflow",
producer_run_ref=SECRET,
idempotency_key=SECRET,
request_hash="d" * 64,
datasource_id=datasource.id,
materialization_id=materialization.id,
status="published",
details_={"secret": SECRET},
created_by="account-1",
),
)
)
@staticmethod
def _datasource(row_id: str, tenant_id: str) -> DatasourceRecord:
return DatasourceRecord(
id=row_id,
tenant_id=tenant_id,
source_name=f"source-{row_id}",
name=SECRET,
description=SECRET,
kind="table",
mode="cached",
shape="tabular",
status="active",
provider="connector.provider",
provider_ref=SECRET,
schema_version=1,
schema_=[{"secret": SECRET}],
fingerprint="e" * 64,
row_count=1,
byte_count=100,
provenance_={"secret": SECRET},
metadata_={"secret": SECRET},
owner_ref=SECRET,
steward_ref=SECRET,
authoritative_source_ref=SECRET,
authority_mode="external_mirror",
legal_basis_refs=[SECRET],
purposes=[SECRET],
semantic_definition=SECRET,
official_keys=[SECRET],
classification="personal",
privacy_profile_ref=SECRET,
retention_policy_ref=SECRET,
hold_refs=[SECRET],
publication_state="published",
transfer_agreement_ref=SECRET,
freshness_policy={"secret": SECRET},
quality_policy={"secret": SECRET},
known_limits=[SECRET],
correction_procedure_ref=SECRET,
affected_refs=[SECRET],
dependency_refs=[SECRET],
created_by="account-1",
updated_by="account-1",
)
@staticmethod
def _payload(
row_id: str,
tenant_id: str,
*,
created_by: str,
) -> DatasourcePayloadRecord:
return DatasourcePayloadRecord(
id=row_id,
tenant_id=tenant_id,
backend="database_rows",
state="published",
locator=SECRET,
media_type="application/x-ndjson",
checksum="f" * 64,
row_count=1,
byte_count=100,
checkpoint_={"secret": SECRET},
metadata_={"secret": SECRET},
created_by=created_by,
)
@staticmethod
def _stage(
row_id: str,
*,
target_datasource_id: str,
promoted: bool,
) -> DatasourceStageRecord:
return DatasourceStageRecord(
id=row_id,
tenant_id="tenant-1",
target_datasource_id=target_datasource_id,
name=SECRET,
source_name=SECRET,
description=SECRET,
kind="table",
mode="static",
shape="tabular",
state="promoted" if promoted else "ready",
provider="upload",
provider_ref=SECRET,
schema_=[{"secret": SECRET}],
rows=[{"secret": SECRET}],
fingerprint="0" * 64,
row_count=1,
byte_count=100,
validation_={"secret": SECRET},
provenance_={"secret": SECRET},
metadata_={"secret": SECRET},
governance_={"secret": SECRET},
promoted_at=NOW if promoted else None,
promoted_materialization_id="materialization-1" if promoted else None,
created_by="account-1",
)
def test_canonical_selector_exports_only_minimized_attribution(self) -> None:
records = self.provider.search_subject(
self.session,
tenant_id="tenant-1",
subject=DsarSubjectRef(account_id="account-1"),
)
self.assertEqual(7, len(records))
self.assertEqual(
{"datasource_operator_attribution"},
{record.category for record in records},
)
exported = json.dumps([record.to_dict() for record in records])
self.assertNotIn(SECRET, exported)
self.assertNotIn("account-1", exported)
self.assertNotIn("datasource-other", exported)
def test_exact_datasource_returns_minimized_lifecycle_package(self) -> None:
records = self.provider.search_subject(
self.session,
tenant_id="tenant-1",
subject=DsarSubjectRef(
account_id="account-1",
external_references={
"datasources.datasource": "datasource:datasource-1"
},
),
)
self.assertEqual(7, len(records))
self.assertEqual(
{
"datasource",
"datasource_governance_reference",
"datasource_materialization",
"datasource_payload",
"datasource_stage",
"datasource_publication",
},
{record.resource_type for record in records},
)
exported = json.dumps([record.to_dict() for record in records])
self.assertNotIn(SECRET, exported)
self.assertNotIn("payload-orphan", exported)
actions = self.provider.plan_erasure(
self.session,
tenant_id="tenant-1",
subject=DsarSubjectRef(
external_references={"datasources.datasource": "datasource-1"}
),
records=records,
)
self.assertEqual({"manual_review"}, {action.kind for action in actions})
def test_exact_references_conflicts_and_tenants_fail_closed(self) -> None:
references = {
"datasources.governance_reference": "governance-1",
"datasources.materialization": "materialization-1",
"datasources.payload": "payload-referenced",
"datasources.stage": "stage-1",
"datasources.publication": "publication-1",
}
for key, value in references.items():
with self.subTest(key=key):
records = self.provider.search_subject(
self.session,
tenant_id="tenant-1",
subject=DsarSubjectRef(external_references={key: value}),
)
self.assertEqual(1, len(records))
mismatch = self.provider.search_subject(
self.session,
tenant_id="tenant-1",
subject=DsarSubjectRef(
account_id="account-2",
external_references={"datasources.stage": "stage-1"},
),
)
wrong_tenant = self.provider.search_subject(
self.session,
tenant_id="tenant-2",
subject=DsarSubjectRef(
external_references={"datasources.stage": "stage-1"}
),
)
conflict = self.provider.search_subject(
self.session,
tenant_id="tenant-1",
subject=DsarSubjectRef(
external_references={
"datasources.datasource": "datasource-1",
"datasources.catalogue": "different",
}
),
)
self.assertEqual((), mismatch)
self.assertEqual((), wrong_tenant)
self.assertEqual((), conflict)
def test_transient_deletion_is_safe_and_idempotent(self) -> None:
subject = DsarSubjectRef(
external_references={
"datasources.stage": "stage-1",
"datasources.payload": "payload-orphan",
}
)
records = self.provider.search_subject(
self.session,
tenant_id="tenant-1",
subject=subject,
)
actions = self.provider.plan_erasure(
self.session,
tenant_id="tenant-1",
subject=subject,
records=records,
)
self.assertEqual({"delete"}, {action.kind for action in actions})
first = self.provider.execute_erasure(
self.session,
tenant_id="tenant-1",
subject=subject,
actions=actions,
request_id="dsar-1",
)
second = self.provider.execute_erasure(
self.session,
tenant_id="tenant-1",
subject=subject,
actions=actions,
request_id="dsar-1-retry",
)
self.assertTrue(all(result.status == "executed" for result in first))
self.assertTrue(all(result.status == "unchanged" for result in second))
self.assertIsNone(self.session.get(DatasourceStageRecord, "stage-1"))
self.assertIsNone(self.session.get(DatasourcePayloadRecord, "payload-orphan"))
self.assertIsNotNone(
self.session.get(DatasourcePayloadRecord, "payload-referenced")
)
def test_published_and_attribution_state_requires_review_or_retention(self) -> None:
direct_subject = DsarSubjectRef(
external_references={
"datasources.materialization": "materialization-1",
"datasources.payload": "payload-referenced",
"datasources.stage": "stage-promoted",
}
)
direct = self.provider.search_subject(
self.session,
tenant_id="tenant-1",
subject=direct_subject,
)
direct_actions = self.provider.plan_erasure(
self.session,
tenant_id="tenant-1",
subject=direct_subject,
records=direct,
)
self.assertEqual(
{"manual_review"},
{action.kind for action in direct_actions},
)
canonical_subject = DsarSubjectRef(account_id="account-1")
canonical = self.provider.search_subject(
self.session,
tenant_id="tenant-1",
subject=canonical_subject,
)
canonical_actions = self.provider.plan_erasure(
self.session,
tenant_id="tenant-1",
subject=canonical_subject,
records=canonical,
)
self.assertEqual({"retain"}, {action.kind for action in canonical_actions})
results = self.provider.execute_erasure(
self.session,
tenant_id="tenant-1",
subject=canonical_subject,
actions=canonical_actions,
request_id="dsar-2",
)
self.assertTrue(all(result.status == "blocked" for result in results))
def test_foreign_records_and_actions_are_rejected(self) -> None:
subject = DsarSubjectRef(account_id="account-1")
with self.assertRaisesRegex(ValueError, "foreign provider record"):
self.provider.plan_erasure(
self.session,
tenant_id="tenant-1",
subject=subject,
records=(
DsarRecordRef(
provider_id="cases",
module_id="cases",
resource_type="case",
resource_id="case-1",
category="case",
title="Case",
),
),
)
with self.assertRaisesRegex(ValueError, "foreign provider action"):
self.provider.execute_erasure(
self.session,
tenant_id="tenant-1",
subject=subject,
actions=(
DsarErasureActionRef(
action_id="cases:delete:case:case-1",
provider_id="cases",
module_id="cases",
kind="delete",
resource_type="case",
resource_id="case-1",
title="Delete case",
rationale="Foreign",
executable=True,
),
),
request_id="dsar-3",
)
def test_core_workflow_reports_active_and_inactive_provider(self) -> None:
row = create_data_subject_request(
self.session,
tenant_id="tenant-1",
reference="DSAR-DATASOURCES-1",
request_kind="access_and_erasure",
subject=DsarSubjectRef(account_id="account-1"),
purpose="Respond to a verified request.",
legal_basis="Article 15 and 17 GDPR",
due_at=None,
requested_by_account_id="privacy-officer",
)
self.session.commit()
search_data_subject_request(
self.session,
registry=_Registry(self.provider),
row=row,
expected_revision=1,
)
self.assertEqual(
[DATASOURCES_DSAR_CAPABILITY],
row.coverage["provider_capabilities"],
)
self.assertEqual(7, row.search_result["record_count"])
inactive = create_data_subject_request(
self.session,
tenant_id="tenant-1",
reference="DSAR-DATASOURCES-2",
request_kind="access",
subject=DsarSubjectRef(account_id="account-1"),
purpose="Respond to a verified request.",
legal_basis="Article 15 GDPR",
due_at=None,
requested_by_account_id="privacy-officer",
)
self.session.commit()
search_data_subject_request(
self.session,
registry=_Registry(self.provider, active=False),
row=inactive,
expected_revision=1,
)
self.assertEqual([], inactive.coverage["provider_capabilities"])
self.assertEqual(
[DATASOURCES_DSAR_CAPABILITY],
inactive.coverage["inactive_provider_capabilities"],
)
self.assertEqual(0, inactive.search_result["record_count"])
def test_manifest_registers_and_documents_capability(self) -> None:
self.assertIn(DATASOURCES_DSAR_CAPABILITY, manifest.capability_factories)
self.assertIn(
DATASOURCES_DSAR_CAPABILITY,
manifest.capability_documentation,
)
self.assertIn(
DATASOURCES_DSAR_CAPABILITY,
{item.name for item in manifest.provides_interfaces},
)
self.assertTrue(
any(
topic.id == "datasources.data-subject-requests"
and {"admin", "user"}.issubset(topic.documentation_types)
for topic in manifest.documentation
)
)
if __name__ == "__main__":
unittest.main()
+195
View File
@@ -11,6 +11,7 @@ from govoplan_core.core.change_sequence import ChangeSequenceEntry
from govoplan_core.core.datasources import ( from govoplan_core.core.datasources import (
CAPABILITY_DATASOURCE_ORIGINS, CAPABILITY_DATASOURCE_ORIGINS,
DatasourceAccessError, DatasourceAccessError,
DatasourceArtifactReference,
DatasourceField, DatasourceField,
DatasourceGovernance, DatasourceGovernance,
DatasourceOrigin, DatasourceOrigin,
@@ -44,6 +45,7 @@ from govoplan_datasources.backend.service import (
SqlDatasourceProvider, SqlDatasourceProvider,
) )
from govoplan_datasources.backend.payloads import ( from govoplan_datasources.backend.payloads import (
ExternalArtifactPayloadBackend,
create_database_rows_payload, create_database_rows_payload,
finalize_payload_deletion, finalize_payload_deletion,
mark_unreferenced_payload_for_deletion, mark_unreferenced_payload_for_deletion,
@@ -166,6 +168,56 @@ class FakeRegistry:
return self.origin_provider return self.origin_provider
class FakeArtifactBackend:
backend = "test_artifact"
def __init__(self) -> None:
self.verified: list[str] = []
self.deleted: list[str] = []
def read_rows(
self,
_session,
*,
tenant_id: str,
artifact: DatasourceArtifactReference,
offset: int,
limit: int,
):
self.assert_tenant(tenant_id)
stop = min(artifact.row_count, offset + limit)
return tuple(
{"id": index, "result": "match"}
for index in range(offset, stop)
)
def verify(
self,
_session,
*,
tenant_id: str,
artifact: DatasourceArtifactReference,
) -> None:
self.assert_tenant(tenant_id)
if not artifact.locator.startswith("artifact:"):
raise DatasourceUnavailableError("Unknown test artifact.")
self.verified.append(artifact.locator)
def delete(
self,
_session,
*,
tenant_id: str,
artifact: DatasourceArtifactReference,
) -> None:
self.assert_tenant(tenant_id)
self.deleted.append(artifact.locator)
def assert_tenant(self, tenant_id: str) -> None:
if tenant_id != "tenant-1":
raise AssertionError("Artifact backend crossed a tenant boundary.")
class DatasourceLifecycleTests(unittest.TestCase): class DatasourceLifecycleTests(unittest.TestCase):
def setUp(self) -> None: def setUp(self) -> None:
self.engine = create_engine("sqlite:///:memory:") self.engine = create_engine("sqlite:///:memory:")
@@ -185,8 +237,10 @@ class DatasourceLifecycleTests(unittest.TestCase):
self.Session = sessionmaker(bind=self.engine) self.Session = sessionmaker(bind=self.engine)
self.session = self.Session() self.session = self.Session()
self.origins = FakeOriginProvider() self.origins = FakeOriginProvider()
self.artifacts = FakeArtifactBackend()
self.provider = SqlDatasourceProvider( self.provider = SqlDatasourceProvider(
registry=FakeRegistry(self.origins), registry=FakeRegistry(self.origins),
payload_backends=(ExternalArtifactPayloadBackend(self.artifacts),),
) )
def tearDown(self) -> None: def tearDown(self) -> None:
@@ -667,6 +721,147 @@ class DatasourceLifecycleTests(unittest.TestCase):
self.session.query(DatasourcePublicationRecord).count(), self.session.query(DatasourcePublicationRecord).count(),
) )
def test_artifact_publication_pins_large_payload_and_supports_bounded_reads(
self,
) -> None:
artifact = DatasourceArtifactReference(
backend="test_artifact",
locator="artifact:monthly-output",
checksum="a" * 64,
row_count=25_000,
byte_count=12_000_000,
schema=(
DatasourceField("id", "integer", nullable=False),
DatasourceField("result", "string", nullable=False),
),
fingerprint="b" * 64,
)
published = self.provider.publish_rows(
self.session,
principal(scopes=(SOURCE_WRITE_SCOPE, CATALOGUE_READ_SCOPE)),
request=DatasourcePublicationRequest(
producer_module="dataflow",
producer_run_ref="dataflow-run:large-output",
idempotency_key="large-output",
name="Large output",
source_name="large_output",
artifact=artifact,
),
)
preview = self.provider.read_datasource(
self.session,
principal(scopes=(CATALOGUE_READ_SCOPE,)),
request=DatasourceReadRequest(
datasource_ref=published.datasource.ref,
offset=10,
limit=3,
),
)
self.assertEqual("published", published.status)
self.assertEqual(25_000, published.materialization.row_count)
self.assertEqual(
[{"id": 10, "result": "match"},
{"id": 11, "result": "match"},
{"id": 12, "result": "match"}],
list(preview.rows),
)
self.assertEqual(
["artifact:monthly-output", "artifact:monthly-output"],
self.artifacts.verified,
)
def test_unattested_artifact_quality_rules_require_review_without_becoming_current(
self,
) -> None:
published = self.provider.publish_rows(
self.session,
principal(scopes=(SOURCE_WRITE_SCOPE,)),
request=DatasourcePublicationRequest(
producer_module="reporting",
producer_run_ref="report-run:review",
idempotency_key="review-output",
name="Review output",
source_name="review_output",
artifact=DatasourceArtifactReference(
backend="test_artifact",
locator="artifact:review-output",
checksum="c" * 64,
row_count=2,
byte_count=128,
schema=(DatasourceField("id", "integer", False),),
fingerprint="d" * 64,
),
governance=DatasourceGovernance(
quality_policy={
"version": "unique-id-v1",
"rules": [
{
"id": "unique-id",
"type": "unique",
"fields": ["id"],
}
],
}
),
),
)
record = self.session.get(
DatasourceRecord,
published.datasource.ref.removeprefix("datasource:"),
)
self.assertEqual("review_required", published.status)
self.assertEqual("review_required", published.materialization.state)
self.assertIsNotNone(record)
assert record is not None
self.assertIsNone(record.current_materialization_id)
def test_artifact_warning_is_a_notification_ready_terminal_state(self) -> None:
published = self.provider.publish_rows(
self.session,
principal(scopes=(SOURCE_WRITE_SCOPE,)),
request=DatasourcePublicationRequest(
producer_module="dataflow",
producer_run_ref="dataflow-run:warning",
idempotency_key="warning-output",
name="Warning output",
source_name="warning_output",
artifact=DatasourceArtifactReference(
backend="test_artifact",
locator="artifact:warning-output",
checksum="e" * 64,
row_count=1,
byte_count=64,
schema=(DatasourceField("id", "integer", False),),
fingerprint="f" * 64,
validation={
"status": "warning",
"warnings": [
{
"severity": "warning",
"code": "producer.partial_match",
"message": "One source used a fallback match.",
}
],
},
),
),
)
record = self.session.get(
DatasourcePublicationRecord,
published.ref.removeprefix("publication:"),
)
self.assertEqual("published_with_warnings", published.status)
self.assertIsNotNone(record)
assert record is not None
self.assertEqual(
"published_with_warnings",
record.status,
)
def test_publication_idempotency_key_rejects_different_output(self) -> None: def test_publication_idempotency_key_rejects_different_output(self) -> None:
producer = principal(scopes=(SOURCE_WRITE_SCOPE,)) producer = principal(scopes=(SOURCE_WRITE_SCOPE,))
base = DatasourcePublicationRequest( base = DatasourcePublicationRequest(
+162
View File
@@ -0,0 +1,162 @@
from __future__ import annotations
from types import SimpleNamespace
import unittest
from sqlalchemy import create_engine
from sqlalchemy.orm import Session
from govoplan_core.auth import ApiPrincipal
from govoplan_core.core.access import PrincipalRef
from govoplan_core.core.search import (
SearchAuthorizationRequest,
SearchBackfillRequest,
SearchResourceReference,
)
from govoplan_core.db.base import Base
from govoplan_datasources.backend.db.models import DatasourceRecord
from govoplan_datasources.backend.manifest import get_manifest
from govoplan_datasources.backend.search_source import (
DatasourcesSearchSource,
PROVIDER_ID,
RESOURCE_TYPE,
)
from govoplan_datasources.backend.service import ADMIN_SCOPE, CATALOGUE_READ_SCOPE
class DatasourcesSearchSourceTests(unittest.TestCase):
def setUp(self) -> None:
self.engine = create_engine("sqlite://")
Base.metadata.create_all(
self.engine,
tables=(DatasourceRecord.__table__,),
)
self.session = Session(self.engine)
self.session.add_all(
(
_datasource("source-1", "tenant-1", "Monthly source"),
_datasource("source-other", "tenant-2", "Other tenant"),
)
)
self.session.commit()
self.source = DatasourcesSearchSource()
def tearDown(self) -> None:
self.session.close()
self.engine.dispose()
def test_manifest_registers_optional_search_source(self) -> None:
manifest = get_manifest()
self.assertIn("search", manifest.optional_dependencies)
self.assertIn(
PROVIDER_ID,
{registration.id for registration in manifest.search_sources},
)
def test_backfill_is_tenant_bound_and_excludes_protected_payloads(self) -> None:
page = self.source.backfill(
self.session,
request=SearchBackfillRequest(
tenant_id="tenant-1",
provider_id=PROVIDER_ID,
resource_type=RESOURCE_TYPE,
rebuild_id="rebuild-1",
),
)
self.assertEqual(("source-1",), tuple(doc.resource_id for doc in page.documents))
document = page.documents[0]
serialized = repr(document)
self.assertNotIn("must-not-be-indexed", serialized)
self.assertNotIn("secret_field", serialized)
self.assertNotIn("credential-1", serialized)
self.assertTrue(document.requires_authorization_recheck)
self.assertEqual(
"/datasources?datasource=datasource%3Asource-1",
document.url,
)
def test_authorization_rechecks_scope_tenant_and_current_existence(self) -> None:
reference = SearchResourceReference(
tenant_id="tenant-1",
module_id="datasources",
resource_type=RESOURCE_TYPE,
resource_id="source-1",
)
request = SearchAuthorizationRequest(reference=reference, source_revision="1")
self.assertTrue(
self.source.authorize(
self.session,
_principal({CATALOGUE_READ_SCOPE}),
requests=(request,),
)[reference.key]
)
self.assertTrue(
self.source.authorize(
self.session,
_principal({ADMIN_SCOPE}),
requests=(request,),
)[reference.key]
)
self.assertFalse(
self.source.authorize(
self.session,
_principal(set()),
requests=(request,),
)[reference.key]
)
other_reference = SearchResourceReference(
tenant_id="tenant-2",
module_id="datasources",
resource_type=RESOURCE_TYPE,
resource_id="source-other",
)
other_request = SearchAuthorizationRequest(
reference=other_reference,
source_revision="1",
)
self.assertFalse(
self.source.authorize(
self.session,
_principal({CATALOGUE_READ_SCOPE}),
requests=(other_request,),
)[other_reference.key]
)
def _datasource(identifier: str, tenant_id: str, name: str) -> DatasourceRecord:
return DatasourceRecord(
id=identifier,
tenant_id=tenant_id,
source_name=identifier.replace("-", "_"),
name=name,
description="Safe catalogue description",
kind="database",
mode="cached",
shape="tabular",
status="active",
provider="connectors.sql",
provider_ref="credential-1",
schema_=[{"name": "secret_field", "type": "string"}],
provenance_={"query": "must-not-be-indexed"},
metadata_={"password": "must-not-be-indexed"},
)
def _principal(scopes: set[str]) -> ApiPrincipal:
return ApiPrincipal(
principal=PrincipalRef(
account_id="account-1",
membership_id="user-1",
tenant_id="tenant-1",
scopes=frozenset(scopes),
),
account=SimpleNamespace(id="account-1"),
user=SimpleNamespace(id="user-1"),
)
if __name__ == "__main__":
unittest.main()
+1 -1
View File
@@ -1,6 +1,6 @@
{ {
"name": "@govoplan/datasources-webui", "name": "@govoplan/datasources-webui",
"version": "0.1.18", "version": "0.1.19",
"private": true, "private": true,
"type": "module", "type": "module",
"main": "src/index.ts", "main": "src/index.ts",
+156 -157
View File
@@ -13,24 +13,34 @@ import {
Layers3, Layers3,
Pencil, Pencil,
Plus, Plus,
RefreshCw,
Snowflake, Snowflake,
ShieldCheck, ShieldCheck,
Trash2, Trash2,
Upload Upload
} from "lucide-react"; } from "lucide-react";
import { import { FormGrid, DialogSection, ActionToolbar,
ActionBlockerHint, ActionBlockerHint,
Button, Button,
ConfirmDialog, ConfirmDialog,
ContentSection,
Dialog, Dialog,
DocumentationHelpLink, DocumentationHelpLink,
DismissibleAlert, DismissibleAlert,
FilterBar,
FormField, FormField,
IconButton, IconButton,
LoadingFrame, LoadingFrame,
MetricCard,
MetricGrid,
SegmentedControl, SegmentedControl,
SelectionList,
SelectionListItem,
SelectionListItemContent,
StatePanel,
StatusBadge, StatusBadge,
WorkspaceActionBar,
WorkspaceFrame,
WorkspaceLayout,
hasScope, hasScope,
isApiError, isApiError,
useUnsavedChanges, useUnsavedChanges,
@@ -83,7 +93,9 @@ export default function DatasourcesPage({
const [stages, setStages] = useState<DatasourceStage[]>([]); const [stages, setStages] = useState<DatasourceStage[]>([]);
const [origins, setOrigins] = useState<DatasourceOrigin[]>([]); const [origins, setOrigins] = useState<DatasourceOrigin[]>([]);
const [originsAvailable, setOriginsAvailable] = useState(false); const [originsAvailable, setOriginsAvailable] = useState(false);
const [selectedDatasourceRef, setSelectedDatasourceRef] = useState(""); const [selectedDatasourceRef, setSelectedDatasourceRef] = useState(
initialDatasourceRef
);
const [selectedStageRef, setSelectedStageRef] = useState(""); const [selectedStageRef, setSelectedStageRef] = useState("");
const [selectedOriginRef, setSelectedOriginRef] = useState(""); const [selectedOriginRef, setSelectedOriginRef] = useState("");
const [preview, setPreview] = useState<DatasourcePreview | null>(null); const [preview, setPreview] = useState<DatasourcePreview | null>(null);
@@ -282,30 +294,32 @@ export default function DatasourcesPage({
}; };
return ( return (
<main className="datasources-page"> <WorkspaceFrame as="main" height="viewport" surface="plain" className="datasources-page" label="Datasource workspace">
<div className="datasources-shell"> <WorkspaceLayout
<aside className="datasources-sidebar" aria-label="Datasource catalogue"> variant="split"
<div className="datasources-sidebar-toolbar"> primarySize="compact"
<strong>Data</strong> surface="contained"
<span className="datasources-toolbar-actions"> primaryScrollable={false}
<IconButton contentScrollable={false}
label="Refresh" primaryLabel="Datasource catalogue"
icon={<RefreshCw size={16} />} contentLabel="Datasource workspace"
variant="ghost" contentClassName="datasources-workspace"
onClick={() => void reload(selectedDatasourceRef)} primary={<>
disabled={loading || working} <WorkspaceActionBar
disabledReason={loading ? DATASOURCES_I18N.loading : working ? DATASOURCES_I18N.working : undefined} scope="collection-pane"
/> variant="collection"
<IconButton refreshable
reloadAction={{ onReload: () => void reload(selectedDatasourceRef), loading: loading || working }}
contextActions={<strong>Data</strong>}
createAction={<IconButton
label="Add datasource or stage" label="Add datasource or stage"
icon={<Plus size={17} />} icon={<Plus size={17} />}
variant="primary" variant="primary"
onClick={() => setAddOpen(true)} onClick={() => setAddOpen(true)}
disabled={!canStage && !canManage} disabled={!canStage && !canManage}
disabledReason={!canStage && !canManage ? DATASOURCES_I18N.manageReason : undefined} disabledReason={!canStage && !canManage ? DATASOURCES_I18N.manageReason : undefined}
/>}
/> />
</span>
</div>
<div className="datasources-view-switch"> <div className="datasources-view-switch">
<SegmentedControl<CatalogueView> <SegmentedControl<CatalogueView>
ariaLabel="Datasource view" ariaLabel="Datasource view"
@@ -324,7 +338,7 @@ export default function DatasourcesPage({
}} }}
/> />
</div> </div>
<div className="datasources-search"> <FilterBar surface="panel">
<input <input
type="search" type="search"
value={search} value={search}
@@ -332,72 +346,60 @@ export default function DatasourcesPage({
placeholder={`Search ${view}`} placeholder={`Search ${view}`}
aria-label={`Search ${view}`} aria-label={`Search ${view}`}
/> />
</div> </FilterBar>
<LoadingFrame <LoadingFrame
loading={loading} loading={loading}
className="datasources-list-frame" className="datasources-list-frame"
> >
<div className="datasources-list"> <SelectionList variant="navigation" label="Datasources">
{view === "catalogue" ? visibleDatasources.map((item) => ( {view === "catalogue" ? visibleDatasources.map((item) => (
<button <SelectionListItem
key={item.ref} key={item.ref}
type="button" selected={item.ref === selectedDatasourceRef}
className={item.ref === selectedDatasourceRef ? "is-selected" : ""}
onClick={() => setSelectedDatasourceRef(item.ref)} onClick={() => setSelectedDatasourceRef(item.ref)}
> >
<span> <SelectionListItemContent title={item.name} description={`${item.source_name} · ${item.kind}`} />
<strong>{item.name}</strong>
<small>{item.source_name} · {item.kind}</small>
</span>
<StatusBadge status={item.mode} label={item.mode} /> <StatusBadge status={item.mode} label={item.mode} />
</button> </SelectionListItem>
)) : null} )) : null}
{view === "staging" ? visibleStages.map((item) => ( {view === "staging" ? visibleStages.map((item) => (
<button <SelectionListItem
key={item.ref} key={item.ref}
type="button" selected={item.ref === selectedStageRef}
className={item.ref === selectedStageRef ? "is-selected" : ""}
onClick={() => setSelectedStageRef(item.ref)} onClick={() => setSelectedStageRef(item.ref)}
> >
<span> <SelectionListItemContent title={item.name} description={`${item.source_name} · ${formatCount(item.row_count, "row")}`} />
<strong>{item.name}</strong>
<small>{item.source_name} · {formatCount(item.row_count, "row")}</small>
</span>
<StatusBadge status={item.state} label={item.state} /> <StatusBadge status={item.state} label={item.state} />
</button> </SelectionListItem>
)) : null} )) : null}
{view === "origins" ? visibleOrigins.map((item) => ( {view === "origins" ? visibleOrigins.map((item) => (
<button <SelectionListItem
key={item.ref} key={item.ref}
type="button" selected={item.ref === selectedOriginRef}
className={item.ref === selectedOriginRef ? "is-selected" : ""}
onClick={() => setSelectedOriginRef(item.ref)} onClick={() => setSelectedOriginRef(item.ref)}
> >
<span> <SelectionListItemContent title={item.name} description={`${item.provider} · ${item.kind}`} />
<strong>{item.name}</strong>
<small>{item.provider} · {item.kind}</small>
</span>
<StatusBadge status={item.shape} label={item.shape} /> <StatusBadge status={item.shape} label={item.shape} />
</button> </SelectionListItem>
)) : null} )) : null}
{view === "catalogue" && !visibleDatasources.length ? ( {view === "catalogue" && !visibleDatasources.length ? (
<div className="datasources-list-empty">No datasources</div> <StatePanel size="compact" description="No datasources" />
) : null} ) : null}
{view === "staging" && !visibleStages.length ? ( {view === "staging" && !visibleStages.length ? (
<div className="datasources-list-empty">No staged data</div> <StatePanel size="compact" description="No staged data" />
) : null} ) : null}
{view === "origins" && !visibleOrigins.length ? ( {view === "origins" && !visibleOrigins.length ? (
<div className="datasources-list-empty"> <StatePanel size="compact" description={originsAvailable ? "No connector origins" : "Connectors unavailable"} />
{originsAvailable ? "No connector origins" : "Connectors unavailable"}
</div>
) : null} ) : null}
</div> </SelectionList>
</LoadingFrame> </LoadingFrame>
</aside> </>}
>
<section className="datasources-workspace"> <WorkspaceActionBar
<div className="datasources-workspace-toolbar"> scope="detail-pane"
<span className="datasources-current-title"> variant="detail"
className="datasources-workspace-toolbar"
contextActions={<span className="datasources-current-title">
{view === "catalogue" ? <DatabaseZap size={19} /> {view === "catalogue" ? <DatabaseZap size={19} />
: view === "staging" ? <Layers3 size={19} /> : view === "staging" ? <Layers3 size={19} />
: <Database size={19} />} : <Database size={19} />}
@@ -417,9 +419,9 @@ export default function DatasourcesPage({
: selectedOrigin?.provider ?? "External acquisition"} : selectedOrigin?.provider ?? "External acquisition"}
</small> </small>
</span> </span>
</span> </span>}
<span className="datasources-toolbar-actions"> helpAction={<DocumentationHelpLink reference={DATASOURCES_DOCUMENTATION} />}
<DocumentationHelpLink reference={DATASOURCES_DOCUMENTATION} /> primaryActions={<>
{view === "catalogue" && selectedDatasource?.mode === "cached" ? ( {view === "catalogue" && selectedDatasource?.mode === "cached" ? (
<Button onClick={() => void refreshSelected()} disabled={!canManage || working} disabledReason={working ? DATASOURCES_I18N.working : !canManage ? DATASOURCES_I18N.manageReason : undefined}> <Button onClick={() => void refreshSelected()} disabled={!canManage || working} disabledReason={working ? DATASOURCES_I18N.working : !canManage ? DATASOURCES_I18N.manageReason : undefined}>
<Download size={16} /> Refresh <Download size={16} /> Refresh
@@ -437,14 +439,6 @@ export default function DatasourcesPage({
<Button onClick={() => setFreezeOpen(true)} disabled={!canManage || working} disabledReason={working ? DATASOURCES_I18N.working : !canManage ? DATASOURCES_I18N.manageReason : undefined}> <Button onClick={() => setFreezeOpen(true)} disabled={!canManage || working} disabledReason={working ? DATASOURCES_I18N.working : !canManage ? DATASOURCES_I18N.manageReason : undefined}>
<Snowflake size={16} /> Freeze <Snowflake size={16} /> Freeze
</Button> </Button>
<IconButton
label="Retire datasource"
icon={<Trash2 size={16} />}
variant="danger"
onClick={() => setRetireOpen(true)}
disabled={!canManage || working}
disabledReason={working ? DATASOURCES_I18N.working : !canManage ? DATASOURCES_I18N.manageReason : undefined}
/>
</> </>
) : null} ) : null}
{view === "staging" && selectedStage?.state === "ready" ? ( {view === "staging" && selectedStage?.state === "ready" ? (
@@ -467,8 +461,18 @@ export default function DatasourcesPage({
<Plus size={16} /> Register <Plus size={16} /> Register
</Button> </Button>
) : null} ) : null}
</span> </>}
</div> destructiveActions={view === "catalogue" && selectedDatasource ? (
<IconButton
label="Retire datasource"
icon={<Trash2 size={16} />}
variant="danger"
onClick={() => setRetireOpen(true)}
disabled={!canManage || working}
disabledReason={working ? DATASOURCES_I18N.working : !canManage ? DATASOURCES_I18N.manageReason : undefined}
/>
) : undefined}
/>
<div className="datasources-alerts"> <div className="datasources-alerts">
{error ? ( {error ? (
@@ -520,8 +524,7 @@ export default function DatasourcesPage({
) : null} ) : null}
</div> </div>
{working ? <div className="datasources-working" role="status">Working...</div> : null} {working ? <div className="datasources-working" role="status">Working...</div> : null}
</section> </WorkspaceLayout>
</div>
<AddDatasourceDialog <AddDatasourceDialog
open={addOpen} open={addOpen}
@@ -605,7 +608,7 @@ export default function DatasourcesPage({
onCancel={() => setRetireOpen(false)} onCancel={() => setRetireOpen(false)}
onConfirm={() => void retireSelected()} onConfirm={() => void retireSelected()}
/> />
</main> </WorkspaceFrame>
); );
} }
@@ -622,24 +625,24 @@ function DatasourceDetail({
}) { }) {
return ( return (
<> <>
<div className="datasources-metrics"> <MetricGrid columns={5} density="compact" minimum="compact">
<Metric label="Mode" value={datasource.mode} /> <MetricCard density="compact" label="Mode" value={datasource.mode} valueTitle={datasource.mode} />
<Metric label="Rows" value={formatNumber(datasource.row_count)} /> <MetricCard density="compact" label="Rows" value={formatNumber(datasource.row_count)} />
<Metric label="Fields" value={String(datasource.schema.length)} /> <MetricCard density="compact" label="Fields" value={String(datasource.schema.length)} />
<Metric label="Revisions" value={String(materializations.length)} /> <MetricCard density="compact" label="Revisions" value={String(materializations.length)} />
<Metric label="Updated" value={formatDate(datasource.updated_at)} /> <MetricCard density="compact" label="Updated" value={formatDate(datasource.updated_at)} />
</div> </MetricGrid>
{datasource.description ? ( {datasource.description ? (
<div className="datasources-description">{datasource.description}</div> <div className="datasources-description">{datasource.description}</div>
) : null} ) : null}
<section className="datasources-detail-section"> <ContentSection>
<div className="datasources-section-heading"> <ActionToolbar surface="section-header" className="datasources-section-heading">
<span><ShieldCheck size={16} /> Governance</span> <span><ShieldCheck size={16} /> Governance</span>
<StatusBadge <StatusBadge
status={datasource.governance.publication_state} status={datasource.governance.publication_state}
label={datasource.governance.publication_state} label={datasource.governance.publication_state}
/> />
</div> </ActionToolbar>
<div className="datasources-key-values"> <div className="datasources-key-values">
<span><small>Authority</small><strong>{readableToken(datasource.governance.authority_mode)}</strong></span> <span><small>Authority</small><strong>{readableToken(datasource.governance.authority_mode)}</strong></span>
<span><small>Classification</small><strong>{datasource.governance.classification}</strong></span> <span><small>Classification</small><strong>{datasource.governance.classification}</strong></span>
@@ -651,25 +654,25 @@ function DatasourceDetail({
{datasource.governance.semantic_definition ? ( {datasource.governance.semantic_definition ? (
<p className="datasources-dialog-copy">{datasource.governance.semantic_definition}</p> <p className="datasources-dialog-copy">{datasource.governance.semantic_definition}</p>
) : null} ) : null}
</section> </ContentSection>
<section className="datasources-detail-section"> <ContentSection>
<div className="datasources-section-heading"> <ActionToolbar surface="section-header" className="datasources-section-heading">
<span><Eye size={16} /> Preview</span> <span><Eye size={16} /> Preview</span>
<small>{preview ? `${formatNumber(preview.total_rows)} total rows` : ""}</small> <small>{preview ? `${formatNumber(preview.total_rows)} total rows` : ""}</small>
</div> </ActionToolbar>
{loading ? ( {loading ? (
<div className="datasources-inline-loading">Loading preview...</div> <div className="datasources-inline-loading">Loading preview...</div>
) : preview ? ( ) : preview ? (
<PreviewTable datasource={datasource} rows={preview.rows} /> <PreviewTable datasource={datasource} rows={preview.rows} />
) : ( ) : (
<div className="datasources-inline-empty">No preview available</div> <StatePanel size="inline" description="No preview available" />
)} )}
</section> </ContentSection>
<section className="datasources-detail-section"> <ContentSection>
<div className="datasources-section-heading"> <ActionToolbar surface="section-header" className="datasources-section-heading">
<span><Archive size={16} /> Materializations</span> <span><Archive size={16} /> Materializations</span>
<small>Immutable revisions</small> <small>Immutable revisions</small>
</div> </ActionToolbar>
<div className="datasources-table-scroll"> <div className="datasources-table-scroll">
<table> <table>
<thead> <thead>
@@ -706,14 +709,14 @@ function DatasourceDetail({
</tbody> </tbody>
</table> </table>
</div> </div>
</section> </ContentSection>
<section className="datasources-detail-section"> <ContentSection>
<div className="datasources-section-heading"> <ActionToolbar surface="section-header" className="datasources-section-heading">
<span>Schema</span> <span>Schema</span>
<small>Version {datasource.schema_version}</small> <small>Version {datasource.schema_version}</small>
</div> </ActionToolbar>
<SchemaTable fields={datasource.schema} /> <SchemaTable fields={datasource.schema} />
</section> </ContentSection>
</> </>
); );
} }
@@ -725,21 +728,21 @@ function StageDetail({ stage }: { stage: DatasourceStage }) {
const schemaClassification = stage.validation.schema_change?.classification ?? "new"; const schemaClassification = stage.validation.schema_change?.classification ?? "new";
return ( return (
<> <>
<div className="datasources-metrics"> <MetricGrid columns={5} density="compact" minimum="compact">
<Metric label="State" value={stage.state} /> <MetricCard density="compact" label="State" value={stage.state} valueTitle={stage.state} />
<Metric label="Mode" value={stage.mode} /> <MetricCard density="compact" label="Mode" value={stage.mode} valueTitle={stage.mode} />
<Metric label="Rows" value={formatNumber(stage.row_count)} /> <MetricCard density="compact" label="Rows" value={formatNumber(stage.row_count)} />
<Metric label="Fields" value={String(stage.schema.length)} /> <MetricCard density="compact" label="Fields" value={String(stage.schema.length)} />
<Metric label="Created" value={formatDate(stage.created_at)} /> <MetricCard density="compact" label="Created" value={formatDate(stage.created_at)} />
</div> </MetricGrid>
<section className="datasources-detail-section"> <ContentSection>
<div className="datasources-section-heading"> <ActionToolbar surface="section-header" className="datasources-section-heading">
<span>Validation</span> <span>Validation</span>
<StatusBadge <StatusBadge
status={stage.validation.valid === false ? "invalid" : "ready"} status={stage.validation.valid === false ? "invalid" : "ready"}
label={stage.validation.valid === false ? "Invalid" : "Ready"} label={stage.validation.valid === false ? "Invalid" : "Ready"}
/> />
</div> </ActionToolbar>
<div className="datasources-key-values"> <div className="datasources-key-values">
<span><small>Fingerprint</small><strong>{shortFingerprint(stage.fingerprint)}</strong></span> <span><small>Fingerprint</small><strong>{shortFingerprint(stage.fingerprint)}</strong></span>
<span><small>Target</small><strong>{stage.target_datasource_ref || "New datasource"}</strong></span> <span><small>Target</small><strong>{stage.target_datasource_ref || "New datasource"}</strong></span>
@@ -768,27 +771,27 @@ function StageDetail({ stage }: { stage: DatasourceStage }) {
All configured quality rules passed and no blocking schema change was detected. All configured quality rules passed and no blocking schema change was detected.
</div> </div>
)} )}
</section> </ContentSection>
{schemaChanges.length ? ( {schemaChanges.length ? (
<section className="datasources-detail-section"> <ContentSection>
<div className="datasources-section-heading"> <ActionToolbar surface="section-header" className="datasources-section-heading">
<span>Schema comparison</span> <span>Schema comparison</span>
<StatusBadge status={schemaClassification} label={readableToken(schemaClassification)} /> <StatusBadge status={schemaClassification} label={readableToken(schemaClassification)} />
</div> </ActionToolbar>
<ul className="datasources-schema-changes"> <ul className="datasources-schema-changes">
{schemaChanges.map((change, index) => ( {schemaChanges.map((change, index) => (
<SchemaChangeItem key={`${change.code}-${change.field ?? index}`} change={change} /> <SchemaChangeItem key={`${change.code}-${change.field ?? index}`} change={change} />
))} ))}
</ul> </ul>
</section> </ContentSection>
) : null} ) : null}
<section className="datasources-detail-section"> <ContentSection>
<div className="datasources-section-heading"> <ActionToolbar surface="section-header" className="datasources-section-heading">
<span>Detected schema</span> <span>Detected schema</span>
<small>{stage.shape}</small> <small>{stage.shape}</small>
</div> </ActionToolbar>
<SchemaTable fields={stage.schema} /> <SchemaTable fields={stage.schema} />
</section> </ContentSection>
</> </>
); );
} }
@@ -832,22 +835,22 @@ function SchemaChangeItem({ change }: { change: DatasourceSchemaChange }) {
function OriginDetail({ origin }: { origin: DatasourceOrigin }) { function OriginDetail({ origin }: { origin: DatasourceOrigin }) {
return ( return (
<> <>
<div className="datasources-metrics"> <MetricGrid columns="auto" density="compact" minimum="compact">
<Metric label="Provider" value={origin.provider} /> <MetricCard density="compact" label="Provider" value={origin.provider} valueTitle={origin.provider} />
<Metric label="Mode" value={readableToken(origin.source_mode)} /> <MetricCard density="compact" label="Mode" value={readableToken(origin.source_mode)} />
<Metric label="Health" value={readableToken(origin.health.status)} /> <MetricCard density="compact" label="Health" value={readableToken(origin.health.status)} />
<Metric label="Rows" value={formatNumber(origin.row_count)} /> <MetricCard density="compact" label="Rows" value={formatNumber(origin.row_count)} />
<Metric label="Fields" value={String(origin.schema.length)} /> <MetricCard density="compact" label="Fields" value={String(origin.schema.length)} />
<Metric label="Updated" value={formatDate(origin.updated_at)} /> <MetricCard density="compact" label="Updated" value={formatDate(origin.updated_at)} />
</div> </MetricGrid>
{origin.description ? ( {origin.description ? (
<div className="datasources-description">{origin.description}</div> <div className="datasources-description">{origin.description}</div>
) : null} ) : null}
<section className="datasources-detail-section"> <ContentSection>
<div className="datasources-section-heading"> <ActionToolbar surface="section-header" className="datasources-section-heading">
<span>Registration options</span> <span>Registration options</span>
<small>{origin.supported_modes.join(", ")}</small> <small>{origin.supported_modes.join(", ")}</small>
</div> </ActionToolbar>
<div className="datasources-key-values"> <div className="datasources-key-values">
<span><small>Origin reference</small><strong>{origin.ref}</strong></span> <span><small>Origin reference</small><strong>{origin.ref}</strong></span>
<span><small>Kind</small><strong>{readableToken(origin.kind)}</strong></span> <span><small>Kind</small><strong>{readableToken(origin.kind)}</strong></span>
@@ -856,14 +859,14 @@ function OriginDetail({ origin }: { origin: DatasourceOrigin }) {
<span><small>Health</small><strong>{origin.health.summary}</strong></span> <span><small>Health</small><strong>{origin.health.summary}</strong></span>
<span><small>Pushdown</small><strong>{pushdownSummary(origin)}</strong></span> <span><small>Pushdown</small><strong>{pushdownSummary(origin)}</strong></span>
</div> </div>
</section> </ContentSection>
<section className="datasources-detail-section"> <ContentSection>
<div className="datasources-section-heading"> <ActionToolbar surface="section-header" className="datasources-section-heading">
<span>Discovered schema</span> <span>Discovered schema</span>
<small>Version {origin.schema_version}</small> <small>Version {origin.schema_version}</small>
</div> </ActionToolbar>
<SchemaTable fields={origin.schema} /> <SchemaTable fields={origin.schema} />
</section> </ContentSection>
</> </>
); );
} }
@@ -970,8 +973,8 @@ function GovernanceDialog({
> >
{error ? <DismissibleAlert tone="danger" resetKey={error}>{error}</DismissibleAlert> : null} {error ? <DismissibleAlert tone="danger" resetKey={error}>{error}</DismissibleAlert> : null}
{draft ? ( {draft ? (
<div className="datasources-dialog-fields"> <DialogSection className="datasources-dialog-fields">
<div className="datasources-dialog-grid"> <FormGrid columns={2} gap="small" collapseAt="narrow">
<FormField label="Authority mode" documentation={DATASOURCE_GOVERNANCE_DOCUMENTATION}> <FormField label="Authority mode" documentation={DATASOURCE_GOVERNANCE_DOCUMENTATION}>
<select <select
value={draft.authority_mode} value={draft.authority_mode}
@@ -1014,20 +1017,20 @@ function GovernanceDialog({
<FormField label="Privacy profile"> <FormField label="Privacy profile">
<input value={draft.privacy_profile_ref ?? ""} onChange={(event) => setValue("privacy_profile_ref", event.target.value || null)} /> <input value={draft.privacy_profile_ref ?? ""} onChange={(event) => setValue("privacy_profile_ref", event.target.value || null)} />
</FormField> </FormField>
<FormField label="Retention policy"> <FormField label="Retention policy" interfaceId="datasources.field.retention-policy" helpContextId="datasources.field.retention-policy" helpModuleId="datasources" documentation={DATASOURCE_GOVERNANCE_DOCUMENTATION}>
<input value={draft.retention_policy_ref ?? ""} onChange={(event) => setValue("retention_policy_ref", event.target.value || null)} /> <input value={draft.retention_policy_ref ?? ""} onChange={(event) => setValue("retention_policy_ref", event.target.value || null)} />
</FormField> </FormField>
<FormField label="Transfer agreement"> <FormField label="Transfer agreement" interfaceId="datasources.field.transfer-agreement" helpContextId="datasources.field.transfer-agreement" helpModuleId="datasources" documentation={DATASOURCE_GOVERNANCE_DOCUMENTATION}>
<input value={draft.transfer_agreement_ref ?? ""} onChange={(event) => setValue("transfer_agreement_ref", event.target.value || null)} /> <input value={draft.transfer_agreement_ref ?? ""} onChange={(event) => setValue("transfer_agreement_ref", event.target.value || null)} />
</FormField> </FormField>
<FormField label="Correction procedure"> <FormField label="Correction procedure">
<input value={draft.correction_procedure_ref ?? ""} onChange={(event) => setValue("correction_procedure_ref", event.target.value || null)} /> <input value={draft.correction_procedure_ref ?? ""} onChange={(event) => setValue("correction_procedure_ref", event.target.value || null)} />
</FormField> </FormField>
</div> </FormGrid>
<FormField label="Semantic definition" documentation={DATASOURCE_GOVERNANCE_DOCUMENTATION}> <FormField label="Semantic definition" documentation={DATASOURCE_GOVERNANCE_DOCUMENTATION}>
<textarea value={draft.semantic_definition ?? ""} onChange={(event) => setValue("semantic_definition", event.target.value || null)} /> <textarea value={draft.semantic_definition ?? ""} onChange={(event) => setValue("semantic_definition", event.target.value || null)} />
</FormField> </FormField>
<div className="datasources-dialog-grid"> <FormGrid columns={2} gap="small" collapseAt="narrow">
<GovernanceListField label="Purposes" values={draft.purposes} onChange={(values) => setValue("purposes", values)} /> <GovernanceListField label="Purposes" values={draft.purposes} onChange={(values) => setValue("purposes", values)} />
<GovernanceListField label="Legal basis references" values={draft.legal_basis_refs} onChange={(values) => setValue("legal_basis_refs", values)} /> <GovernanceListField label="Legal basis references" values={draft.legal_basis_refs} onChange={(values) => setValue("legal_basis_refs", values)} />
<GovernanceListField label="Official keys" values={draft.official_keys} onChange={(values) => setValue("official_keys", values)} /> <GovernanceListField label="Official keys" values={draft.official_keys} onChange={(values) => setValue("official_keys", values)} />
@@ -1035,16 +1038,16 @@ function GovernanceDialog({
<GovernanceListField label="Affected services and processes" values={draft.affected_refs} onChange={(values) => setValue("affected_refs", values)} /> <GovernanceListField label="Affected services and processes" values={draft.affected_refs} onChange={(values) => setValue("affected_refs", values)} />
<GovernanceListField label="Dependent flows, reports, controls and decisions" values={draft.dependency_refs} onChange={(values) => setValue("dependency_refs", values)} /> <GovernanceListField label="Dependent flows, reports, controls and decisions" values={draft.dependency_refs} onChange={(values) => setValue("dependency_refs", values)} />
<GovernanceListField label="Known limits" values={draft.known_limits} onChange={(values) => setValue("known_limits", values)} /> <GovernanceListField label="Known limits" values={draft.known_limits} onChange={(values) => setValue("known_limits", values)} />
</div> </FormGrid>
<div className="datasources-dialog-grid"> <FormGrid columns={2} gap="small" collapseAt="narrow">
<FormField label="Freshness policy (JSON)" documentation={DATASOURCE_GOVERNANCE_DOCUMENTATION}> <FormField label="Freshness policy (JSON)" documentation={DATASOURCE_GOVERNANCE_DOCUMENTATION}>
<textarea value={freshness} onChange={(event) => setFreshness(event.target.value)} spellCheck={false} /> <textarea value={freshness} onChange={(event) => setFreshness(event.target.value)} spellCheck={false} />
</FormField> </FormField>
<FormField label="Quality policy (JSON)" documentation={DATASOURCE_GOVERNANCE_DOCUMENTATION}> <FormField label="Quality policy (JSON)" documentation={DATASOURCE_GOVERNANCE_DOCUMENTATION}>
<textarea value={quality} onChange={(event) => setQuality(event.target.value)} spellCheck={false} /> <textarea value={quality} onChange={(event) => setQuality(event.target.value)} spellCheck={false} />
</FormField> </FormField>
</div> </FormGrid>
</div> </DialogSection>
) : null} ) : null}
</Dialog> </Dialog>
); );
@@ -1281,7 +1284,7 @@ function AddDatasourceDialog({
</> </>
)} )}
> >
<div className="datasources-dialog-fields"> <DialogSection className="datasources-dialog-fields">
{error ? ( {error ? (
<DismissibleAlert tone="danger" resetKey={error}>{error}</DismissibleAlert> <DismissibleAlert tone="danger" resetKey={error}>{error}</DismissibleAlert>
) : null} ) : null}
@@ -1334,7 +1337,7 @@ function AddDatasourceDialog({
</FormField> </FormField>
</> </>
)} )}
<div className="datasources-dialog-grid"> <FormGrid columns={2} gap="small" collapseAt="narrow">
<FormField label="Name" documentation={DATASOURCE_FIELDS_DOCUMENTATION}> <FormField label="Name" documentation={DATASOURCE_FIELDS_DOCUMENTATION}>
<input value={name} onChange={(event) => setName(event.target.value)} /> <input value={name} onChange={(event) => setName(event.target.value)} />
</FormField> </FormField>
@@ -1347,7 +1350,7 @@ function AddDatasourceDialog({
disabled={Boolean(targetRef)} disabled={Boolean(targetRef)}
/> />
</FormField> </FormField>
</div> </FormGrid>
<FormField label="Description"> <FormField label="Description">
<input value={description} onChange={(event) => setDescription(event.target.value)} /> <input value={description} onChange={(event) => setDescription(event.target.value)} />
</FormField> </FormField>
@@ -1419,7 +1422,7 @@ function AddDatasourceDialog({
)} )}
</> </>
) : null} ) : null}
</div> </DialogSection>
</Dialog> </Dialog>
); );
} }
@@ -1477,17 +1480,8 @@ function SchemaTable({ fields }: { fields: Datasource["schema"] }) {
); );
} }
function Metric({ label, value }: { label: string; value: string }) {
return (
<span>
<small>{label}</small>
<strong title={value}>{value}</strong>
</span>
);
}
function EmptyWorkspace({ icon, label }: { icon: React.ReactNode; label: string }) { function EmptyWorkspace({ icon, label }: { icon: React.ReactNode; label: string }) {
return <div className="datasources-workspace-empty">{icon}<strong>{label}</strong></div>; return <StatePanel size="fill" icon={icon} title={label} />;
} }
function filterItems<T>( function filterItems<T>(
@@ -1501,6 +1495,11 @@ function filterItems<T>(
: items; : items;
} }
function initialDatasourceRef(): string {
if (typeof window === "undefined") return "";
return new URLSearchParams(window.location.search).get("datasource") ?? "";
}
function parseRows(text: string): Record<string, unknown>[] { function parseRows(text: string): Record<string, unknown>[] {
const value: unknown = JSON.parse(text); const value: unknown = JSON.parse(text);
if (!Array.isArray(value) || value.some((row) => !isRecord(row))) { if (!Array.isArray(value) || value.some((row) => !isRecord(row))) {
+3 -2
View File
@@ -10,14 +10,15 @@ const readScopes = ["datasources:catalogue:read", "datasources:source:admin"];
export const datasourcesModule: PlatformWebModule = { export const datasourcesModule: PlatformWebModule = {
id: "datasources", id: "datasources",
label: "i18n:govoplan-datasources.datasources", label: "i18n:govoplan-datasources.datasources",
version: "0.1.14", version: "0.1.18",
optionalDependencies: [ optionalDependencies: [
"access", "access",
"audit", "audit",
"connectors", "connectors",
"files", "files",
"notifications", "notifications",
"policy" "policy",
"search"
], ],
translations: generatedTranslations, translations: generatedTranslations,
viewSurfaces: [ viewSurfaces: [
+2 -209
View File
@@ -1,65 +1,9 @@
.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-content,
.datasources-list-frame { .datasources-list-frame {
min-width: 0; min-width: 0;
min-height: 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-toolbar-actions,
.datasources-current-title, .datasources-current-title,
.datasources-current-title > span, .datasources-current-title > span,
@@ -75,80 +19,11 @@
background: var(--panel); 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 { .datasources-list-frame {
flex: 1 1 auto; flex: 1 1 auto;
overflow: hidden; 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 { .datasources-inline-loading {
display: grid; display: grid;
place-items: center; place-items: center;
@@ -158,17 +33,8 @@
text-align: center; text-align: center;
} }
.datasources-workspace {
position: relative;
display: flex;
flex-direction: column;
overflow: hidden;
background: var(--bg);
}
.datasources-workspace-toolbar { .datasources-workspace-toolbar {
min-height: 58px; min-width: 0;
padding: 8px 10px 8px 14px;
} }
.datasources-current-title { .datasources-current-title {
@@ -218,43 +84,6 @@
padding: 14px; 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 { .datasources-description {
margin-bottom: 14px; margin-bottom: 14px;
padding: 10px 12px; padding: 10px 12px;
@@ -265,18 +94,6 @@
line-height: 1.45; 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 { .datasources-section-heading > span {
color: var(--text-strong); color: var(--text-strong);
font-size: 13px; font-size: 13px;
@@ -487,12 +304,6 @@
font-size: 12px; font-size: 12px;
} }
.datasources-dialog-grid {
display: grid;
grid-template-columns: minmax(0, 1fr) minmax(0, 1fr);
gap: 12px;
}
.datasources-dialog-copy { .datasources-dialog-copy {
margin: 0 0 14px; margin: 0 0 14px;
color: var(--muted); color: var(--muted);
@@ -507,26 +318,10 @@
overflow: auto; 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 { .datasources-workspace {
min-height: 58vh; min-height: 58vh;
} }
.datasources-metrics {
grid-template-columns: repeat(2, minmax(0, 1fr));
}
.datasources-workspace-toolbar { .datasources-workspace-toolbar {
align-items: flex-start; align-items: flex-start;
flex-wrap: wrap; flex-wrap: wrap;
@@ -538,9 +333,7 @@
} }
@media (max-width: 560px) { @media (max-width: 560px) {
.datasources-metrics, .datasources-key-values {
.datasources-key-values,
.datasources-dialog-grid {
grid-template-columns: 1fr; grid-template-columns: 1fr;
} }