diff --git a/README.md b/README.md index 4eee001..a899269 100644 --- a/README.md +++ b/README.md @@ -23,6 +23,12 @@ Producer modules can append a bounded tabular result or create a new static datasource through an idempotent capability. The publication ledger retains the producer run, output materialization, provenance, and replay identity. +Tabular staging also evaluates governed quality rules and classifies schema +changes before promotion. Blocking stages remain inspectable, and successful +promotion preserves the exact policy hash and validation result in immutable +materialization provenance. The supported contract is documented in +[docs/QUALITY_POLICY.md](docs/QUALITY_POLICY.md). + The contracts already model database, HTTP/REST, directory, file, feed, document, binary, directory, and stream sources so providers can be added without changing consumers. Larger durable artifact-backed publications remain diff --git a/docs/CONCEPT.md b/docs/CONCEPT.md index e5833b8..cfb0d03 100644 --- a/docs/CONCEPT.md +++ b/docs/CONCEPT.md @@ -76,6 +76,14 @@ A stage records schema, row and byte counts, fingerprint, validation result, provenance, and intended target. Promotion either creates a datasource or adds a new immutable materialization to an existing compatible datasource. +For bounded tabular stages, the Datasource quality policy now enforces row, +field, nullability, uniqueness, numeric range, and embedded referential-set +rules. Updates receive a deterministic compatible, warning, or breaking schema +classification. Errors and breaking changes keep a stage inspectable but block +promotion; warnings remain visible and promotable. The policy hash, diagnostics, +and schema diff are retained with the promoted materialization evidence. See +[QUALITY_POLICY.md](QUALITY_POLICY.md) for the contract and its privacy bounds. + The first slice stores bounded tabular JSON/CSV stages. Future providers may stage file references, object-store blobs, directory snapshots, or streaming checkpoints through the same lifecycle contract. diff --git a/docs/INTERFACE_PATTERN_MIGRATION.md b/docs/INTERFACE_PATTERN_MIGRATION.md index 9a2eca1..f61e326 100644 --- a/docs/INTERFACE_PATTERN_MIGRATION.md +++ b/docs/INTERFACE_PATTERN_MIGRATION.md @@ -9,7 +9,7 @@ editor, previews, and immutable materialization history. | Surface | Archetype | Consequence class | Contract | | --- | --- | --- | --- | | `/datasources` catalogue | Governed directory | Select, register, refresh, freeze, govern, or retire | Shared loading/empty/error, permission, disabled-reason, contextual-help, and read-only states | -| Staging | Review/preflight queue | Upload, validate, inspect, and promote | Non-consumable bounded stage, explicit promotion confirmation, immutable resulting revision | +| Staging | Review/preflight queue | Upload, validate, inspect, and promote | Non-consumable bounded stage with privacy-safe quality/schema diagnostics, explicit promotion confirmation, and immutable validation evidence on the resulting revision | | Connector origins | Optional-provider directory | Register a live or cached source | Provider availability and supported modes without hard Connectors dependency | | Governance editor | Effective authority/provenance editor | Change institutional data context | Guarded draft, authority/source/owner/purpose/quality/freshness semantics | | Preview/materializations | Evidence register | Inspect current sample and immutable revisions | Row/schema bounds, freshness, provenance, fingerprints, hashes, and frozen labels | diff --git a/docs/QUALITY_POLICY.md b/docs/QUALITY_POLICY.md new file mode 100644 index 0000000..25bbe3a --- /dev/null +++ b/docs/QUALITY_POLICY.md @@ -0,0 +1,83 @@ +# Datasource Stage Quality Policy + +Datasource quality policy is a deterministic JSON contract stored in +`governance.quality_policy`. A tabular stage inherits the current target +Datasource policy unless the stage has its own governed definition. Validation +runs before the stage is stored, but a failed stage remains available for +inspection and correction through a new stage. + +## Contract + +```json +{ + "version": "monthly-import-v1", + "rules": [ + {"id": "non-empty", "type": "row_count", "minimum": 1}, + {"id": "columns", "type": "required_fields", "fields": ["id", "status"]}, + {"id": "id-shape", "type": "field", "field": "id", "data_type": "integer", "nullable": false}, + {"id": "id-present", "type": "not_null", "fields": ["id"]}, + {"id": "id-unique", "type": "unique", "fields": ["id"]}, + {"id": "amount-range", "type": "range", "field": "amount", "minimum": 0}, + {"id": "known-status", "type": "referential", "field": "status", "allowed_values": ["new", "closed"]} + ], + "schema_policy": { + "field_added_required": "warning", + "field_removed": "breaking" + } +} +``` + +Every rule has a stable `id` and may set `severity` to `error` or `warning`. +Errors block promotion; warnings require review but leave the stage ready. The +supported rules are: + +- `row_count`: optional non-negative `minimum` and `maximum`; +- `required_fields`: fields that must exist in the detected schema; +- `field`: expected `data_type` and/or `nullable` contract for one field; +- `not_null`: one `field` or a `fields` list that must contain no nulls; +- `unique`: one `field` or a composite `fields` list, with optional + `ignore_nulls`; +- `range`: numeric `minimum` and/or `maximum`, with optional `allow_null`; +- `referential`: a field and a bounded `allowed_values` set, with optional + `allow_null`. + +The bounded referential rule is intentionally local and reproducible. A future +cross-Datasource reference rule must freeze the referenced materialization and +perform an independent read-authorization check; it must not silently read the +current state of another protected Datasource. + +Use embedded values only for non-sensitive code sets. Quality policy is +catalogue governance metadata and can be visible to actors who are not allowed +to read protected rows. + +## Schema Classification + +When a stage targets an existing Datasource, fields are compared by stable +name. Defaults are: + +| Change | Default | +| --- | --- | +| nullable field added | compatible | +| required field added | warning | +| field removed | breaking | +| integer widened to number, or unknown resolved | compatible | +| other type change | breaking | +| nullability relaxed | breaking | +| nullability tightened | compatible | +| existing field order changed | warning | + +Each key may be overridden in `schema_policy` with `compatible`, `warning`, or +`breaking`. The highest resulting classification is the stage classification. +A breaking classification blocks promotion. + +## Evidence And Privacy + +Validation records the policy contract version, a SHA-256 hash of the complete +policy, rule counts, diagnostics, and the complete schema diff. Row diagnostics +contain only affected counts and at most 25 one-based row numbers; they never +copy field values. Promotion copies this validation object into the immutable +materialization provenance and records the policy hash in the audit event. + +Approval authority, approval expiry, and retention/deletion execution remain +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. diff --git a/src/govoplan_datasources/backend/manifest.py b/src/govoplan_datasources/backend/manifest.py index 9e9d626..64cb954 100644 --- a/src/govoplan_datasources/backend/manifest.py +++ b/src/govoplan_datasources/backend/manifest.py @@ -17,6 +17,7 @@ from govoplan_core.core.module_guards import ( persistent_table_uninstall_guard, ) from govoplan_core.core.modules import ( + DocumentationLink, DocumentationTopic, FrontendModule, FrontendRoute, @@ -381,7 +382,8 @@ manifest = ModuleManifest( "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 " "known limits provide discoverable institutional context. Freshness and quality policies are typed JSON contracts retained " - "with materialization evidence; enforcement remains with the provider or consuming control that declares support. Metadata " + "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 " "visibility never grants row access." ), layer="available", @@ -402,6 +404,44 @@ manifest = ModuleManifest( ], }, ), + DocumentationTopic( + id="datasources.quality-gates", + title="Validate a Datasource stage", + summary="Apply deterministic quality rules and schema-change policy before staged data can become consumable.", + body=( + "Tabular stages evaluate configured row count, required-field, field-shape, nullability, uniqueness, numeric-range, and bounded " + "referential-set rules. Errors keep the stage inspectable but block promotion; warnings stay visible and permit an explicit " + "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 " + "schema diff are copied into immutable materialization provenance when promotion succeeds. Approval and retention execution are " + "not inferred from arbitrary JSON flags and remain separate governed lifecycle work." + ), + layer="available", + documentation_types=("admin", "user"), + audience=("operator", "module_admin", "data_steward", "power_user"), + links=( + DocumentationLink( + label="Datasource quality policy", + href="govoplan-datasources/docs/QUALITY_POLICY.md", + kind="repository", + ), + ), + related_modules=("policy", "approvals", "audit", "dataflow", "workflow_engine"), + order=72, + metadata={ + "seed": True, + "help_contexts": [ + "datasources.staging", + "datasources.staging.validation", + "datasources.field.quality-policy", + "datasources.action.promote", + ], + "limitations": [ + "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.", + ], + }, + ), DocumentationTopic( id="datasources.reference.fields-and-consequences", title="Datasource fields and lifecycle consequences", @@ -419,7 +459,7 @@ manifest = ModuleManifest( documentation_types=("admin", "user"), audience=("operator", "module_admin", "power_user", "product_owner"), related_modules=("connectors", "dataflow", "workflow_engine", "reporting", "audit"), - order=72, + order=73, metadata={ "seed": True, "help_contexts": [ diff --git a/src/govoplan_datasources/backend/quality.py b/src/govoplan_datasources/backend/quality.py new file mode 100644 index 0000000..93c8cf8 --- /dev/null +++ b/src/govoplan_datasources/backend/quality.py @@ -0,0 +1,654 @@ +from __future__ import annotations + +import hashlib +import json +from collections.abc import Mapping, Sequence +from dataclasses import dataclass, field +from typing import Literal + +from govoplan_core.core.datasources import DatasourceField + + +DiagnosticSeverity = Literal["error", "warning"] +SchemaClassification = Literal["compatible", "warning", "breaking"] + +QUALITY_POLICY_VERSION = 1 +MAX_DIAGNOSTIC_ROWS = 25 +MAX_REFERENCE_VALUES = 10_000 + + +@dataclass(frozen=True, slots=True) +class QualityDiagnostic: + severity: DiagnosticSeverity + code: str + message: str + rule_id: str | None = None + affected_rows: int | None = None + row_numbers: tuple[int, ...] = () + details: Mapping[str, object] = field(default_factory=dict) + + def to_dict(self) -> dict[str, object]: + payload: dict[str, object] = { + "severity": self.severity, + "code": self.code, + "message": self.message, + } + if self.rule_id: + payload["rule_id"] = self.rule_id + if self.affected_rows is not None: + payload["affected_rows"] = self.affected_rows + if self.row_numbers: + payload["row_numbers"] = list(self.row_numbers) + payload["row_numbers_truncated"] = self.affected_rows is not None and self.affected_rows > len(self.row_numbers) + if self.details: + payload["details"] = dict(self.details) + return payload + + +@dataclass(frozen=True, slots=True) +class SchemaChange: + code: str + classification: SchemaClassification + message: str + field_name: str | None = None + before: Mapping[str, object] | None = None + after: Mapping[str, object] | None = None + + def to_dict(self) -> dict[str, object]: + payload: dict[str, object] = { + "code": self.code, + "classification": self.classification, + "message": self.message, + } + if self.field_name: + payload["field"] = self.field_name + if self.before is not None: + payload["before"] = dict(self.before) + if self.after is not None: + payload["after"] = dict(self.after) + return payload + + +_SCHEMA_DEFAULTS: dict[str, SchemaClassification] = { + "field_added_nullable": "compatible", + "field_added_required": "warning", + "field_removed": "breaking", + "type_widened": "compatible", + "type_changed": "breaking", + "nullability_relaxed": "breaking", + "nullability_tightened": "compatible", + "field_order_changed": "warning", +} +_CLASSIFICATION_RANK = {"compatible": 0, "warning": 1, "breaking": 2} + + +def validate_stage( + *, + rows: Sequence[Mapping[str, object]], + schema: Sequence[DatasourceField], + quality_policy: Mapping[str, object] | None, + baseline_schema: Sequence[DatasourceField] | None = None, +) -> dict[str, object]: + policy = dict(quality_policy or {}) + diagnostics: list[QualityDiagnostic] = [] + evaluated_rule_ids: list[str] = [] + failed_rule_ids: set[str] = set() + + raw_rules = policy.get("rules", []) + if not _is_sequence(raw_rules): + diagnostics.append(_policy_error("Quality policy rules must be a list.")) + raw_rules = [] + + known_rule_ids: set[str] = set() + for index, raw_rule in enumerate(raw_rules): + fallback_rule_id = f"rule-{index + 1}" + if not isinstance(raw_rule, Mapping): + diagnostics.append(_policy_error("Each quality rule must be an object.", fallback_rule_id)) + failed_rule_ids.add(fallback_rule_id) + continue + rule_id = str(raw_rule.get("id") or fallback_rule_id).strip() + if not rule_id: + rule_id = fallback_rule_id + if rule_id in known_rule_ids: + diagnostics.append(_policy_error(f"Quality rule id {rule_id!r} is duplicated.", rule_id)) + failed_rule_ids.add(rule_id) + continue + known_rule_ids.add(rule_id) + evaluated_rule_ids.append(rule_id) + rule_diagnostics = _evaluate_rule( + rule_id=rule_id, + rule=raw_rule, + rows=rows, + schema=schema, + ) + diagnostics.extend(rule_diagnostics) + if rule_diagnostics: + failed_rule_ids.add(rule_id) + + schema_policy, schema_policy_diagnostics = _schema_policy(policy.get("schema_policy")) + diagnostics.extend(schema_policy_diagnostics) + schema_changes = ( + _compare_schemas(baseline_schema, schema, schema_policy) + if baseline_schema is not None + else () + ) + schema_classification: str = "new" + if baseline_schema is not None: + schema_classification = _aggregate_schema_classification(schema_changes) + for change in schema_changes: + if change.classification == "warning": + diagnostics.append( + QualityDiagnostic( + severity="warning", + code=change.code, + message=change.message, + details={"field": change.field_name} if change.field_name else {}, + ) + ) + elif change.classification == "breaking": + diagnostics.append( + QualityDiagnostic( + severity="error", + code=change.code, + message=change.message, + details={"field": change.field_name} if change.field_name else {}, + ) + ) + + errors = [item.to_dict() for item in diagnostics if item.severity == "error"] + warnings = [item.to_dict() for item in diagnostics if item.severity == "warning"] + return { + "version": QUALITY_POLICY_VERSION, + "policy_version": str(policy.get("version") or QUALITY_POLICY_VERSION), + "policy_hash": _policy_hash(policy), + "valid": not errors, + "errors": errors, + "warnings": warnings, + "quality": { + "rules_evaluated": len(evaluated_rule_ids), + "rules_passed": len(evaluated_rule_ids) - len(failed_rule_ids.intersection(evaluated_rule_ids)), + "rules_failed": len(failed_rule_ids), + }, + "schema_change": { + "classification": schema_classification, + "changes": [change.to_dict() for change in schema_changes], + }, + } + + +def _evaluate_rule( + *, + rule_id: str, + rule: Mapping[str, object], + rows: Sequence[Mapping[str, object]], + schema: Sequence[DatasourceField], +) -> tuple[QualityDiagnostic, ...]: + kind = str(rule.get("type") or "").strip() + severity = str(rule.get("severity") or "error").strip() + if severity not in {"error", "warning"}: + return (_policy_error("Quality rule severity must be error or warning.", rule_id),) + if kind == "row_count": + return _row_count_rule(rule_id, severity, rule, len(rows)) + if kind == "required_fields": + return _required_fields_rule(rule_id, severity, rule, schema) + if kind == "field": + return _field_rule(rule_id, severity, rule, schema) + if kind == "not_null": + return _not_null_rule(rule_id, severity, rule, rows, schema) + if kind == "unique": + return _unique_rule(rule_id, severity, rule, rows, schema) + if kind == "range": + return _range_rule(rule_id, severity, rule, rows, schema) + if kind == "referential": + return _referential_rule(rule_id, severity, rule, rows, schema) + return (_policy_error(f"Unsupported quality rule type {kind!r}.", rule_id),) + + +def _row_count_rule( + rule_id: str, + severity: str, + rule: Mapping[str, object], + row_count: int, +) -> tuple[QualityDiagnostic, ...]: + minimum = _optional_non_negative_int(rule.get("minimum")) + maximum = _optional_non_negative_int(rule.get("maximum")) + if rule.get("minimum") is not None and minimum is None: + return (_policy_error("A row_count rule minimum must be a non-negative integer.", rule_id),) + if rule.get("maximum") is not None and maximum is None: + return (_policy_error("A row_count rule maximum must be a non-negative integer.", rule_id),) + if minimum is None and maximum is None: + return (_policy_error("A row_count rule requires minimum or maximum.", rule_id),) + if minimum is not None and maximum is not None and minimum > maximum: + return (_policy_error("A row_count rule minimum cannot exceed maximum.", rule_id),) + if (minimum is not None and row_count < minimum) or (maximum is not None and row_count > maximum): + return ( + QualityDiagnostic( + severity=_severity(severity), + code="quality.row_count", + message=f"Row count {row_count} is outside the configured bounds.", + rule_id=rule_id, + details={"actual": row_count, "minimum": minimum, "maximum": maximum}, + ), + ) + return () + + +def _required_fields_rule( + rule_id: str, + severity: str, + rule: Mapping[str, object], + schema: Sequence[DatasourceField], +) -> tuple[QualityDiagnostic, ...]: + fields, error = _fields_option(rule, "fields") + if error: + return (_policy_error(error, rule_id),) + known = {field.name for field in schema} + missing = [name for name in fields if name not in known] + if not missing: + return () + return ( + QualityDiagnostic( + severity=_severity(severity), + code="quality.required_fields", + message=f"Required fields are missing: {', '.join(missing)}.", + rule_id=rule_id, + details={"fields": missing}, + ), + ) + + +def _field_rule( + rule_id: str, + severity: str, + rule: Mapping[str, object], + schema: Sequence[DatasourceField], +) -> tuple[QualityDiagnostic, ...]: + field_name = _field_option(rule) + if not field_name: + return (_policy_error("A field rule requires field.", rule_id),) + field_value = next((item for item in schema if item.name == field_name), None) + if field_value is None: + return ( + QualityDiagnostic( + severity=_severity(severity), + code="quality.field_missing", + message=f"Required field {field_name!r} is missing.", + rule_id=rule_id, + details={"field": field_name}, + ), + ) + diagnostics: list[QualityDiagnostic] = [] + expected_type = str(rule.get("data_type") or "").strip() + expected_nullable = rule.get("nullable") + if not expected_type and expected_nullable is None: + return (_policy_error("A field rule requires data_type or nullable.", rule_id),) + if expected_type and field_value.data_type != expected_type: + diagnostics.append( + QualityDiagnostic( + severity=_severity(severity), + code="quality.field_type", + message=f"Field {field_name!r} has type {field_value.data_type!r}; expected {expected_type!r}.", + rule_id=rule_id, + details={"field": field_name, "actual": field_value.data_type, "expected": expected_type}, + ) + ) + if expected_nullable is not None and not isinstance(expected_nullable, bool): + diagnostics.append(_policy_error("A field rule nullable value must be boolean.", rule_id)) + elif isinstance(expected_nullable, bool) and field_value.nullable != expected_nullable: + diagnostics.append( + QualityDiagnostic( + severity=_severity(severity), + code="quality.field_nullability", + message=f"Field {field_name!r} nullability does not match the configured contract.", + rule_id=rule_id, + details={"field": field_name, "actual": field_value.nullable, "expected": expected_nullable}, + ) + ) + return tuple(diagnostics) + + +def _not_null_rule( + rule_id: str, + severity: str, + rule: Mapping[str, object], + rows: Sequence[Mapping[str, object]], + schema: Sequence[DatasourceField], +) -> tuple[QualityDiagnostic, ...]: + fields, error = _fields_option(rule, "fields", allow_single_field=True) + if error: + return (_policy_error(error, rule_id),) + missing = _missing_fields(fields, schema) + if missing: + return (_policy_error(f"not_null references unknown fields: {', '.join(missing)}.", rule_id),) + affected = [index + 1 for index, row in enumerate(rows) if any(row.get(name) is None for name in fields)] + return _row_diagnostic( + severity, + "quality.not_null", + f"Rows contain null values in required fields: {', '.join(fields)}.", + rule_id, + affected, + {"fields": list(fields)}, + ) + + +def _unique_rule( + rule_id: str, + severity: str, + rule: Mapping[str, object], + rows: Sequence[Mapping[str, object]], + schema: Sequence[DatasourceField], +) -> tuple[QualityDiagnostic, ...]: + fields, error = _fields_option(rule, "fields", allow_single_field=True) + if error: + return (_policy_error(error, rule_id),) + missing = _missing_fields(fields, schema) + if missing: + return (_policy_error(f"unique references unknown fields: {', '.join(missing)}.", rule_id),) + ignore_nulls = rule.get("ignore_nulls", False) + if not isinstance(ignore_nulls, bool): + return (_policy_error("A unique rule ignore_nulls value must be boolean.", rule_id),) + occurrences: dict[str, list[int]] = {} + for index, row in enumerate(rows, start=1): + values = tuple(row.get(name) for name in fields) + if ignore_nulls and any(value is None for value in values): + continue + key = json.dumps(values, sort_keys=True, separators=(",", ":"), default=str) + occurrences.setdefault(key, []).append(index) + affected = sorted(index for indexes in occurrences.values() if len(indexes) > 1 for index in indexes) + return _row_diagnostic( + severity, + "quality.unique", + f"Rows violate uniqueness for fields: {', '.join(fields)}.", + rule_id, + affected, + {"fields": list(fields)}, + ) + + +def _range_rule( + rule_id: str, + severity: str, + rule: Mapping[str, object], + rows: Sequence[Mapping[str, object]], + schema: Sequence[DatasourceField], +) -> tuple[QualityDiagnostic, ...]: + field_name = _field_option(rule) + if not field_name: + return (_policy_error("A range rule requires field.", rule_id),) + if field_name in _missing_fields((field_name,), schema): + return (_policy_error(f"range references unknown field {field_name!r}.", rule_id),) + minimum = _optional_number(rule.get("minimum")) + maximum = _optional_number(rule.get("maximum")) + if rule.get("minimum") is not None and minimum is None: + return (_policy_error("A range rule minimum must be numeric.", rule_id),) + if rule.get("maximum") is not None and maximum is None: + return (_policy_error("A range rule maximum must be numeric.", rule_id),) + if minimum is None and maximum is None: + return (_policy_error("A range rule requires minimum or maximum.", rule_id),) + if minimum is not None and maximum is not None and minimum > maximum: + return (_policy_error("A range rule minimum cannot exceed maximum.", rule_id),) + allow_null = rule.get("allow_null", True) + if not isinstance(allow_null, bool): + return (_policy_error("A range rule allow_null value must be boolean.", rule_id),) + affected: list[int] = [] + for index, row in enumerate(rows, start=1): + value = row.get(field_name) + if value is None and allow_null: + continue + if not _is_number(value) or (minimum is not None and value < minimum) or (maximum is not None and value > maximum): + affected.append(index) + return _row_diagnostic( + severity, + "quality.range", + f"Rows contain values outside the configured range for {field_name!r}.", + rule_id, + affected, + {"field": field_name, "minimum": minimum, "maximum": maximum}, + ) + + +def _referential_rule( + rule_id: str, + severity: str, + rule: Mapping[str, object], + rows: Sequence[Mapping[str, object]], + schema: Sequence[DatasourceField], +) -> tuple[QualityDiagnostic, ...]: + field_name = _field_option(rule) + if not field_name: + return (_policy_error("A referential rule requires field.", rule_id),) + if field_name in _missing_fields((field_name,), schema): + return (_policy_error(f"referential references unknown field {field_name!r}.", rule_id),) + allowed_values = rule.get("allowed_values") + if not _is_sequence(allowed_values): + return (_policy_error("A bounded referential rule requires allowed_values.", rule_id),) + if len(allowed_values) > MAX_REFERENCE_VALUES: + return (_policy_error(f"A bounded referential rule supports at most {MAX_REFERENCE_VALUES:,} values.", rule_id),) + allow_null = rule.get("allow_null", False) + if not isinstance(allow_null, bool): + return (_policy_error("A referential rule allow_null value must be boolean.", rule_id),) + allowed = {_canonical_value(value) for value in allowed_values} + affected = [ + index + for index, row in enumerate(rows, start=1) + if not (allow_null and row.get(field_name) is None) + and _canonical_value(row.get(field_name)) not in allowed + ] + return _row_diagnostic( + severity, + "quality.referential", + f"Rows contain unrecognized references in {field_name!r}.", + rule_id, + affected, + {"field": field_name, "allowed_value_count": len(allowed)}, + ) + + +def _compare_schemas( + baseline: Sequence[DatasourceField], + candidate: Sequence[DatasourceField], + policy: Mapping[str, SchemaClassification], +) -> tuple[SchemaChange, ...]: + before_by_name = {field.name: field for field in baseline} + after_by_name = {field.name: field for field in candidate} + changes: list[SchemaChange] = [] + for field_name, before in before_by_name.items(): + after = after_by_name.get(field_name) + if after is None: + changes.append(_schema_change("field_removed", field_name, before, None, policy)) + continue + if before.data_type != after.data_type: + kind = "type_widened" if (before.data_type, after.data_type) in {("integer", "number"), ("unknown", after.data_type)} else "type_changed" + changes.append(_schema_change(kind, field_name, before, after, policy)) + if before.nullable != after.nullable: + kind = "nullability_relaxed" if not before.nullable and after.nullable else "nullability_tightened" + changes.append(_schema_change(kind, field_name, before, after, policy)) + for field_name, after in after_by_name.items(): + if field_name in before_by_name: + continue + kind = "field_added_nullable" if after.nullable else "field_added_required" + changes.append(_schema_change(kind, field_name, None, after, policy)) + shared_before = [field.name for field in baseline if field.name in after_by_name] + shared_after = [field.name for field in candidate if field.name in before_by_name] + if shared_before != shared_after: + changes.append( + SchemaChange( + code="schema.field_order_changed", + classification=policy["field_order_changed"], + message="The order of existing fields changed.", + ) + ) + return tuple(changes) + + +def _schema_change( + kind: str, + field_name: str, + before: DatasourceField | None, + after: DatasourceField | None, + policy: Mapping[str, SchemaClassification], +) -> SchemaChange: + before_type = before.data_type if before is not None else "unknown" + after_type = after.data_type if after is not None else "unknown" + messages = { + "field_added_nullable": f"Nullable field {field_name!r} was added.", + "field_added_required": f"Required field {field_name!r} was added.", + "field_removed": f"Field {field_name!r} was removed.", + "type_widened": f"Field {field_name!r} widened from {before_type!r} to {after_type!r}.", + "type_changed": f"Field {field_name!r} changed type from {before_type!r} to {after_type!r}.", + "nullability_relaxed": f"Field {field_name!r} now permits null values.", + "nullability_tightened": f"Field {field_name!r} no longer contains null values.", + } + return SchemaChange( + code=f"schema.{kind}", + classification=policy[kind], + message=messages[kind], + field_name=field_name, + before=_field_dict(before), + after=_field_dict(after), + ) + + +def _schema_policy( + raw_policy: object, +) -> tuple[dict[str, SchemaClassification], tuple[QualityDiagnostic, ...]]: + result = dict(_SCHEMA_DEFAULTS) + if raw_policy is None: + return result, () + if not isinstance(raw_policy, Mapping): + return result, (_policy_error("schema_policy must be an object."),) + diagnostics: list[QualityDiagnostic] = [] + for key, raw_value in raw_policy.items(): + if key not in result: + diagnostics.append(_policy_error(f"Unsupported schema policy change {key!r}.")) + continue + value = str(raw_value).strip() + if value not in _CLASSIFICATION_RANK: + diagnostics.append(_policy_error(f"Schema policy {key!r} must be compatible, warning, or breaking.")) + continue + result[str(key)] = value # type: ignore[assignment] + return result, tuple(diagnostics) + + +def _aggregate_schema_classification(changes: Sequence[SchemaChange]) -> SchemaClassification: + if not changes: + return "compatible" + return max((change.classification for change in changes), key=_CLASSIFICATION_RANK.__getitem__) + + +def _row_diagnostic( + severity: str, + code: str, + message: str, + rule_id: str, + affected: Sequence[int], + details: Mapping[str, object], +) -> tuple[QualityDiagnostic, ...]: + if not affected: + return () + return ( + QualityDiagnostic( + severity=_severity(severity), + code=code, + message=message, + rule_id=rule_id, + affected_rows=len(affected), + row_numbers=tuple(affected[:MAX_DIAGNOSTIC_ROWS]), + details=details, + ), + ) + + +def _policy_error(message: str, rule_id: str | None = None) -> QualityDiagnostic: + return QualityDiagnostic( + severity="error", + code="quality.policy_invalid", + message=message, + rule_id=rule_id, + ) + + +def _fields_option( + rule: Mapping[str, object], + key: str, + *, + allow_single_field: bool = False, +) -> tuple[tuple[str, ...], str | None]: + raw_fields = rule.get(key) + if raw_fields is None and allow_single_field: + raw_fields = [rule.get("field")] if rule.get("field") is not None else None + if not _is_sequence(raw_fields): + return (), f"Quality rule {key} must be a non-empty list." + fields = tuple(str(value).strip() for value in raw_fields) + if not fields or any(not value for value in fields): + return (), f"Quality rule {key} must be a non-empty list." + if len(fields) != len(set(fields)): + return (), f"Quality rule {key} cannot contain duplicates." + return fields, None + + +def _field_option(rule: Mapping[str, object]) -> str: + return str(rule.get("field") or "").strip() + + +def _missing_fields( + fields: Sequence[str], + schema: Sequence[DatasourceField], +) -> tuple[str, ...]: + known = {field.name for field in schema} + return tuple(name for name in fields if name not in known) + + +def _field_dict(value: DatasourceField | None) -> dict[str, object] | None: + if value is None: + return None + return { + "name": value.name, + "data_type": value.data_type, + "nullable": value.nullable, + } + + +def _policy_hash(policy: Mapping[str, object]) -> str: + encoded = json.dumps(policy, sort_keys=True, separators=(",", ":"), default=str) + return hashlib.sha256(encoded.encode("utf-8")).hexdigest() + + +def _canonical_value(value: object) -> str: + return json.dumps(value, sort_keys=True, separators=(",", ":"), default=str) + + +def _optional_non_negative_int(value: object) -> int | None: + if value is None: + return None + if isinstance(value, bool) or not isinstance(value, int) or value < 0: + return None + return value + + +def _optional_number(value: object) -> int | float | None: + if value is None: + return None + return value if _is_number(value) else None + + +def _is_number(value: object) -> bool: + return isinstance(value, (int, float)) and not isinstance(value, bool) + + +def _is_sequence(value: object) -> bool: + return isinstance(value, Sequence) and not isinstance(value, (str, bytes, bytearray)) + + +def _severity(value: str) -> DiagnosticSeverity: + return "warning" if value == "warning" else "error" + + +__all__ = [ + "MAX_DIAGNOSTIC_ROWS", + "MAX_REFERENCE_VALUES", + "QUALITY_POLICY_VERSION", + "QualityDiagnostic", + "SchemaChange", + "validate_stage", +] diff --git a/src/govoplan_datasources/backend/router.py b/src/govoplan_datasources/backend/router.py index 5454b76..a26da21 100644 --- a/src/govoplan_datasources/backend/router.py +++ b/src/govoplan_datasources/backend/router.py @@ -1,5 +1,7 @@ from __future__ import annotations +from collections.abc import Mapping + from fastapi import APIRouter, Depends, HTTPException, Query, status from sqlalchemy.orm import Session @@ -224,6 +226,11 @@ def api_create_stage( "mode": stage.mode, "row_count": stage.row_count, "fingerprint": stage.fingerprint, + "validation_valid": stage.validation.get("valid"), + "quality_policy_hash": stage.validation.get("policy_hash"), + "schema_classification": _safe_mapping( + stage.validation.get("schema_change") + ).get("classification"), }, ) session.commit() @@ -262,6 +269,9 @@ def api_promote_stage( "materialization_ref": materialization.ref, "revision": materialization.revision, "frozen": materialization.frozen_at is not None, + "quality_policy_hash": _safe_mapping( + materialization.provenance.get("stage_validation") + ).get("policy_hash"), }, ) session.commit() @@ -663,6 +673,10 @@ def _governance( return DatasourceGovernance.from_mapping(payload.model_dump()) +def _safe_mapping(value: object) -> Mapping[str, object]: + return value if isinstance(value, Mapping) else {} + + def _audit( session: Session, principal: ApiPrincipal, diff --git a/src/govoplan_datasources/backend/schemas.py b/src/govoplan_datasources/backend/schemas.py index f99744e..6e06cab 100644 --- a/src/govoplan_datasources/backend/schemas.py +++ b/src/govoplan_datasources/backend/schemas.py @@ -121,6 +121,48 @@ class DatasourceMaterializationListResponse(BaseModel): materializations: list[DatasourceMaterializationResponse] +class DatasourceValidationDiagnosticResponse(BaseModel): + severity: Literal["error", "warning"] + code: str + message: str + rule_id: str | None = None + affected_rows: int | None = None + row_numbers: list[int] = Field(default_factory=list) + row_numbers_truncated: bool = False + details: dict[str, Any] = Field(default_factory=dict) + + +class DatasourceSchemaChangeResponse(BaseModel): + code: str + classification: Literal["compatible", "warning", "breaking"] + message: str + field: str | None = None + before: DatasourceFieldResponse | None = None + after: DatasourceFieldResponse | None = None + + +class DatasourceQualitySummaryResponse(BaseModel): + rules_evaluated: int = 0 + rules_passed: int = 0 + rules_failed: int = 0 + + +class DatasourceSchemaChangeSummaryResponse(BaseModel): + classification: Literal["new", "compatible", "warning", "breaking"] = "new" + changes: list[DatasourceSchemaChangeResponse] = Field(default_factory=list) + + +class DatasourceStageValidationResponse(BaseModel): + version: int = 1 + policy_version: str = "1" + policy_hash: str = "" + valid: bool = True + errors: list[DatasourceValidationDiagnosticResponse] = Field(default_factory=list) + warnings: list[DatasourceValidationDiagnosticResponse] = Field(default_factory=list) + quality: DatasourceQualitySummaryResponse = Field(default_factory=DatasourceQualitySummaryResponse) + schema_change: DatasourceSchemaChangeSummaryResponse = Field(default_factory=DatasourceSchemaChangeSummaryResponse) + + class DatasourceStageResponse(BaseModel): ref: str name: str @@ -134,7 +176,7 @@ class DatasourceStageResponse(BaseModel): fields: list[DatasourceFieldResponse] = Field(alias="schema") row_count: int | None byte_count: int | None - validation: dict[str, Any] + validation: DatasourceStageValidationResponse created_at: str | None promoted_at: str | None promoted_materialization_ref: str | None diff --git a/src/govoplan_datasources/backend/service.py b/src/govoplan_datasources/backend/service.py index d2178a1..945da97 100644 --- a/src/govoplan_datasources/backend/service.py +++ b/src/govoplan_datasources/backend/service.py @@ -48,6 +48,7 @@ from govoplan_datasources.backend.payloads import ( payload_for_materialization, validate_payload_size, ) +from govoplan_datasources.backend.quality import validate_stage from govoplan_datasources.backend.tabular import ( MAX_READ_ROWS, MAX_STAGE_ROWS, @@ -427,6 +428,12 @@ class SqlDatasourceProvider: rows = normalize_rows(stage.rows) schema = infer_schema(rows) fingerprint = fingerprint_rows(rows, schema) + validation = validate_stage( + rows=rows, + schema=schema, + quality_policy=governance.quality_policy, + baseline_schema=_fields(target.schema_) if target is not None else None, + ) item = DatasourceStageRecord( tenant_id=api_principal.tenant_id, target_datasource_id=target.id if target else None, @@ -436,7 +443,7 @@ class SqlDatasourceProvider: kind=stage.kind, mode=stage.mode, shape=stage.shape, - state="ready", + state="ready" if validation["valid"] else "invalid", provider=_clean_optional(stage.provider), provider_ref=_clean_optional(stage.provider_ref), schema_=[field_payload(field) for field in schema], @@ -444,7 +451,7 @@ class SqlDatasourceProvider: fingerprint=fingerprint, row_count=len(rows), byte_count=encoded_size(rows), - validation_={"valid": True, "errors": [], "warnings": []}, + validation_=validation, provenance_=dict(stage.provenance), metadata_=dict(stage.metadata), governance_=governance.to_dict(), @@ -469,6 +476,10 @@ class SqlDatasourceProvider: tenant_id=api_principal.tenant_id, stage_ref=stage_ref, ) + if stage.validation_.get("valid") is not True: + raise DatasourceValidationError( + "The stage has blocking quality or schema diagnostics and cannot be promoted." + ) if stage.state != "ready": raise DatasourceValidationError("Only ready stages can be promoted.") datasource = ( @@ -538,6 +549,7 @@ class SqlDatasourceProvider: provenance={ **dict(stage.provenance_), "stage_ref": _stage_ref(stage.id), + "stage_validation": dict(stage.validation_), }, metadata=dict(stage.metadata_), set_current=True, diff --git a/tests/test_interface_documentation_contract.py b/tests/test_interface_documentation_contract.py index 0c9b3f8..b050977 100644 --- a/tests/test_interface_documentation_contract.py +++ b/tests/test_interface_documentation_contract.py @@ -26,10 +26,14 @@ class DatasourcesInterfaceDocumentationContractTests(unittest.TestCase): topics = {topic.id: topic for topic in manifest.documentation} lifecycle = topics["datasources.lifecycle"] governance = topics["datasources.governance"] + quality = topics["datasources.quality-gates"] reference = topics["datasources.reference.fields-and-consequences"] self.assertIn("datasources.staging", lifecycle.metadata["help_contexts"]) self.assertIn("datasources.field.authority-mode", governance.metadata["help_contexts"]) + self.assertIn("datasources.staging.validation", quality.metadata["help_contexts"]) + self.assertIn("policy version and hash", quality.body) + self.assertTrue(quality.metadata["limitations"]) self.assertIn("datasources.action.promote", reference.metadata["help_contexts"]) self.assertIn("freeze", reference.metadata["consequence_classes"]) self.assertIn("retire", reference.metadata["consequence_classes"]) diff --git a/tests/test_lifecycle.py b/tests/test_lifecycle.py index 46f2547..1e7099c 100644 --- a/tests/test_lifecycle.py +++ b/tests/test_lifecycle.py @@ -265,6 +265,96 @@ class DatasourceLifecycleTests(unittest.TestCase): self.assertEqual(first_record.payload_id, frozen_record.payload_id) self.assertEqual([], first_record.rows) + def test_stage_quality_and_schema_gates_block_only_error_diagnostics(self) -> None: + governance = DatasourceGovernance( + authority_mode="native_authoritative", + publication_state="internal", + quality_policy={ + "version": "monthly-cases-v1", + "rules": [ + {"id": "unique-id", "type": "unique", "fields": ["id"]}, + ], + "schema_policy": {"field_added_required": "warning"}, + }, + ) + first_stage = self.provider.create_stage( + self.session, + principal(), + stage=DatasourceStageInput( + name="Monthly cases", + source_name="monthly_quality_cases", + kind="upload", + mode="static", + shape="tabular", + rows=({"id": 1},), + governance=governance, + ), + ) + datasource, _first = self.provider.promote_stage( + self.session, + principal(), + stage_ref=first_stage.ref, + ) + + blocked_stage = self.provider.create_stage( + self.session, + principal(), + stage=DatasourceStageInput( + name="Monthly cases", + source_name="monthly_quality_cases", + kind="upload", + mode="static", + shape="tabular", + target_datasource_ref=datasource.ref, + rows=({"id": 2, "name": "Ada"}, {"id": 2, "name": "Lin"}), + ), + ) + self.assertEqual("invalid", blocked_stage.state) + self.assertFalse(blocked_stage.validation["valid"]) + self.assertEqual( + "quality.unique", + blocked_stage.validation["errors"][0]["code"], + ) + self.assertEqual( + "warning", + blocked_stage.validation["schema_change"]["classification"], + ) + with self.assertRaisesRegex( + DatasourceValidationError, + "blocking quality or schema diagnostics", + ): + self.provider.promote_stage( + self.session, + principal(), + stage_ref=blocked_stage.ref, + ) + + ready_stage = self.provider.create_stage( + self.session, + principal(), + stage=DatasourceStageInput( + name="Monthly cases", + source_name="monthly_quality_cases", + kind="upload", + mode="static", + shape="tabular", + target_datasource_ref=datasource.ref, + rows=({"id": 2, "name": "Ada"}, {"id": 3, "name": "Lin"}), + ), + ) + self.assertEqual("ready", ready_stage.state) + self.assertTrue(ready_stage.validation["valid"]) + self.assertEqual("schema.field_added_required", ready_stage.validation["warnings"][0]["code"]) + _updated, materialization = self.provider.promote_stage( + self.session, + principal(), + stage_ref=ready_stage.ref, + ) + self.assertEqual( + ready_stage.validation["policy_hash"], + materialization.provenance["stage_validation"]["policy_hash"], + ) + def test_governance_is_queryable_and_snapshotted_per_materialization(self) -> None: stage = self.provider.create_stage( self.session, diff --git a/tests/test_quality.py b/tests/test_quality.py new file mode 100644 index 0000000..9e6ef65 --- /dev/null +++ b/tests/test_quality.py @@ -0,0 +1,142 @@ +from __future__ import annotations + +import unittest + +from govoplan_core.core.datasources import DatasourceField +from govoplan_datasources.backend.quality import MAX_DIAGNOSTIC_ROWS, validate_stage +from govoplan_datasources.backend.schemas import DatasourceStageValidationResponse + + +class DatasourceQualityTests(unittest.TestCase): + def test_quality_rules_report_counts_without_exposing_values(self) -> None: + rows = ( + {"id": 1, "name": "Ada", "amount": 10, "status": "new"}, + {"id": 1, "name": None, "amount": -2, "status": "unknown"}, + ) + schema = ( + DatasourceField("id", "integer", False), + DatasourceField("name", "string", True), + DatasourceField("amount", "integer", False), + DatasourceField("status", "string", False), + ) + result = validate_stage( + rows=rows, + schema=schema, + quality_policy={ + "version": "case-import-v2", + "rules": [ + {"id": "rows", "type": "row_count", "minimum": 3}, + {"id": "columns", "type": "required_fields", "fields": ["id", "missing"]}, + {"id": "id-shape", "type": "field", "field": "id", "data_type": "string"}, + {"id": "names", "type": "not_null", "fields": ["name"]}, + {"id": "ids", "type": "unique", "fields": ["id"]}, + {"id": "amount", "type": "range", "field": "amount", "minimum": 0}, + { + "id": "status", + "type": "referential", + "field": "status", + "allowed_values": ["new", "closed"], + }, + ], + }, + ) + + self.assertFalse(result["valid"]) + self.assertEqual("case-import-v2", result["policy_version"]) + errors = result["errors"] + self.assertEqual( + { + "quality.row_count", + "quality.required_fields", + "quality.field_type", + "quality.not_null", + "quality.unique", + "quality.range", + "quality.referential", + }, + {item["code"] for item in errors}, + ) + unique = next(item for item in errors if item["code"] == "quality.unique") + self.assertEqual(2, unique["affected_rows"]) + self.assertEqual([1, 2], unique["row_numbers"]) + self.assertNotIn("unknown", str(errors)) + contract = DatasourceStageValidationResponse.model_validate(result) + self.assertEqual("case-import-v2", contract.policy_version) + self.assertEqual(7, contract.quality.rules_failed) + + def test_warning_rules_do_not_block_promotion_readiness(self) -> None: + result = validate_stage( + rows=({"id": 1},), + schema=(DatasourceField("id", "integer", False),), + quality_policy={ + "rules": [ + { + "id": "large-batch", + "type": "row_count", + "minimum": 2, + "severity": "warning", + } + ] + }, + ) + + self.assertTrue(result["valid"]) + self.assertEqual([], result["errors"]) + self.assertEqual("quality.row_count", result["warnings"][0]["code"]) + + def test_schema_changes_are_classified_and_can_be_governed(self) -> None: + baseline = ( + DatasourceField("id", "integer", False), + DatasourceField("label", "string", False), + ) + candidate = ( + DatasourceField("id", "number", False), + DatasourceField("note", "string", True), + ) + + blocked = validate_stage( + rows=(), + schema=candidate, + baseline_schema=baseline, + quality_policy={}, + ) + self.assertFalse(blocked["valid"]) + self.assertEqual("breaking", blocked["schema_change"]["classification"]) + self.assertEqual( + {"schema.type_widened", "schema.field_removed", "schema.field_added_nullable"}, + {item["code"] for item in blocked["schema_change"]["changes"]}, + ) + + governed = validate_stage( + rows=(), + schema=candidate, + baseline_schema=baseline, + quality_policy={"schema_policy": {"field_removed": "warning"}}, + ) + self.assertTrue(governed["valid"]) + self.assertEqual("warning", governed["schema_change"]["classification"]) + + def test_malformed_policy_fails_closed_and_hash_is_stable(self) -> None: + policy = {"rules": [{"id": "bad", "type": "unique", "fields": []}]} + first = validate_stage(rows=(), schema=(), quality_policy=policy) + second = validate_stage(rows=(), schema=(), quality_policy=policy) + + self.assertFalse(first["valid"]) + self.assertEqual("quality.policy_invalid", first["errors"][0]["code"]) + self.assertEqual(first["policy_hash"], second["policy_hash"]) + + def test_diagnostic_row_numbers_are_bounded(self) -> None: + rows = tuple({"id": None} for _ in range(MAX_DIAGNOSTIC_ROWS + 10)) + result = validate_stage( + rows=rows, + schema=(DatasourceField("id", "unknown", True),), + quality_policy={"rules": [{"id": "id", "type": "not_null", "field": "id"}]}, + ) + error = result["errors"][0] + self.assertEqual(MAX_DIAGNOSTIC_ROWS + 10, error["affected_rows"]) + self.assertEqual(MAX_DIAGNOSTIC_ROWS, len(error["row_numbers"])) + self.assertTrue(error["row_numbers_truncated"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/webui/src/api/datasources.ts b/webui/src/api/datasources.ts index e586481..506072e 100644 --- a/webui/src/api/datasources.ts +++ b/webui/src/api/datasources.ts @@ -87,6 +87,44 @@ export type DatasourceMaterialization = { governance: DatasourceGovernance; }; +export type DatasourceValidationDiagnostic = { + severity: "error" | "warning"; + code: string; + message: string; + rule_id?: string; + affected_rows?: number; + row_numbers?: number[]; + row_numbers_truncated?: boolean; + details?: Record; +}; + +export type DatasourceSchemaChange = { + code: string; + classification: "compatible" | "warning" | "breaking"; + message: string; + field?: string; + before?: DatasourceField; + after?: DatasourceField; +}; + +export type DatasourceStageValidation = { + version?: number; + policy_version?: string; + policy_hash?: string; + valid?: boolean; + errors?: DatasourceValidationDiagnostic[]; + warnings?: DatasourceValidationDiagnostic[]; + quality?: { + rules_evaluated?: number; + rules_passed?: number; + rules_failed?: number; + }; + schema_change?: { + classification?: "new" | "compatible" | "warning" | "breaking"; + changes?: DatasourceSchemaChange[]; + }; +}; + export type DatasourceStage = { ref: string; name: string; @@ -100,7 +138,7 @@ export type DatasourceStage = { schema: DatasourceField[]; row_count?: number | null; byte_count?: number | null; - validation: Record; + validation: DatasourceStageValidation; created_at?: string | null; promoted_at?: string | null; promoted_materialization_ref?: string | null; diff --git a/webui/src/features/datasources/DatasourcesPage.tsx b/webui/src/features/datasources/DatasourcesPage.tsx index 2b329a7..d9a22dd 100644 --- a/webui/src/features/datasources/DatasourcesPage.tsx +++ b/webui/src/features/datasources/DatasourcesPage.tsx @@ -56,7 +56,9 @@ import { type DatasourceMaterialization, type DatasourceOrigin, type DatasourcePreview, - type DatasourceStage + type DatasourceSchemaChange, + type DatasourceStage, + type DatasourceValidationDiagnostic } from "../../api/datasources"; import { DATASOURCE_FIELDS_DOCUMENTATION, @@ -717,6 +719,10 @@ function DatasourceDetail({ } function StageDetail({ stage }: { stage: DatasourceStage }) { + const errors = stage.validation.errors ?? []; + const warnings = stage.validation.warnings ?? []; + const schemaChanges = stage.validation.schema_change?.changes ?? []; + const schemaClassification = stage.validation.schema_change?.classification ?? "new"; return ( <>
@@ -738,8 +744,44 @@ function StageDetail({ stage }: { stage: DatasourceStage }) { Fingerprint{shortFingerprint(stage.fingerprint)} Target{stage.target_datasource_ref || "New datasource"} Promoted revision{stage.promoted_materialization_ref || "Not promoted"} + Quality policy{stage.validation.policy_version || "Local default"} + Policy hash{shortFingerprint(stage.validation.policy_hash || "")} + Schema change{readableToken(schemaClassification)}
+ {errors.length || warnings.length ? ( +
+ {errors.length ? ( + + Promotion blockers + + + ) : null} + {warnings.length ? ( + + Review warnings + + + ) : null} +
+ ) : ( +
+ All configured quality rules passed and no blocking schema change was detected. +
+ )} + {schemaChanges.length ? ( +
+
+ Schema comparison + +
+
    + {schemaChanges.map((change, index) => ( + + ))} +
+
+ ) : null}
Detected schema @@ -751,6 +793,42 @@ function StageDetail({ stage }: { stage: DatasourceStage }) { ); } +function ValidationDiagnosticList({ + diagnostics +}: { + diagnostics: DatasourceValidationDiagnostic[]; +}) { + return ( +
    + {diagnostics.map((diagnostic, index) => { + const details = [ + diagnostic.rule_id ? `Rule ${diagnostic.rule_id}` : "", + diagnostic.affected_rows !== undefined ? `${diagnostic.affected_rows} affected row${diagnostic.affected_rows === 1 ? "" : "s"}` : "", + diagnostic.row_numbers?.length ? `Rows ${diagnostic.row_numbers.join(", ")}${diagnostic.row_numbers_truncated ? ", …" : ""}` : "" + ].filter(Boolean); + return ( +
  • + {diagnostic.message} + {details.length ? {details.join(" · ")} : null} +
    • + ); + })} +
+ ); +} + +function SchemaChangeItem({ change }: { change: DatasourceSchemaChange }) { + return ( +
  • + + + {change.message} + {change.field ? {change.field} : null} + +
    • + ); +} + function OriginDetail({ origin }: { origin: DatasourceOrigin }) { return ( <> diff --git a/webui/src/i18n/generatedTranslations.ts b/webui/src/i18n/generatedTranslations.ts index 07c0c2a..ef82475 100644 --- a/webui/src/i18n/generatedTranslations.ts +++ b/webui/src/i18n/generatedTranslations.ts @@ -62,7 +62,15 @@ const en = { "Publication state": "Publication state", "Semantic definition": "Semantic definition", "Freshness policy (JSON)": "Freshness policy (JSON)", - "Quality policy (JSON)": "Quality policy (JSON)" + "Quality policy (JSON)": "Quality policy (JSON)", + "Quality policy": "Quality policy", + "Policy hash": "Policy hash", + "Schema change": "Schema change", + "Local default": "Local default", + "Promotion blockers": "Promotion blockers", + "Review warnings": "Review warnings", + "All configured quality rules passed and no blocking schema change was detected.": "All configured quality rules passed and no blocking schema change was detected.", + "Schema comparison": "Schema comparison" } as const; const de: Record = { @@ -127,7 +135,15 @@ const de: Record = { "Publication state": "Veröffentlichungsstatus", "Semantic definition": "Semantische Definition", "Freshness policy (JSON)": "Aktualitätsrichtlinie (JSON)", - "Quality policy (JSON)": "Qualitätsrichtlinie (JSON)" + "Quality policy (JSON)": "Qualitätsrichtlinie (JSON)", + "Quality policy": "Qualitätsrichtlinie", + "Policy hash": "Richtlinien-Hash", + "Schema change": "Schemaänderung", + "Local default": "Lokaler Standard", + "Promotion blockers": "Übernahmehindernisse", + "Review warnings": "Prüfhinweise", + "All configured quality rules passed and no blocking schema change was detected.": "Alle konfigurierten Qualitätsregeln wurden erfüllt und es wurde keine blockierende Schemaänderung erkannt.", + "Schema comparison": "Schemavergleich" }; export const generatedTranslations: PlatformTranslations = { en, de }; diff --git a/webui/src/styles/datasources.css b/webui/src/styles/datasources.css index fd1e710..7afd78c 100644 --- a/webui/src/styles/datasources.css +++ b/webui/src/styles/datasources.css @@ -369,6 +369,74 @@ font-weight: 600; } +.datasources-validation-alerts { + display: grid; + gap: 8px; + padding: 10px; + border-top: var(--border-line); +} + +.datasources-validation-alerts .alert { + margin: 0; +} + +.datasources-diagnostic-list, +.datasources-schema-changes { + display: grid; + gap: 7px; + margin: 7px 0 0; + padding: 0; + list-style: none; +} + +.datasources-diagnostic-list li, +.datasources-diagnostic-list span, +.datasources-diagnostic-list small, +.datasources-schema-changes li, +.datasources-schema-changes span, +.datasources-schema-changes strong, +.datasources-schema-changes small { + display: block; + min-width: 0; +} + +.datasources-diagnostic-list small, +.datasources-schema-changes small { + margin-top: 2px; + color: var(--muted); + font-size: 11px; +} + +.datasources-validation-ok { + padding: 10px 12px; + border-top: var(--border-line); + color: var(--success); + font-size: 12px; +} + +.datasources-schema-changes { + margin: 0; + padding: 8px 10px; +} + +.datasources-schema-changes li { + display: grid; + grid-template-columns: auto minmax(0, 1fr); + align-items: start; + gap: 9px; + padding: 6px 0; +} + +.datasources-schema-changes li + li { + border-top: var(--border-line); +} + +.datasources-schema-changes strong { + color: var(--text-strong); + font-size: 12px; + font-weight: 600; +} + .datasources-workspace-empty { display: grid; place-items: center;