feat: implement governed reporting vertical
This commit is contained in:
@@ -0,0 +1,976 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Mapping, Sequence
|
||||
from datetime import UTC, date, datetime
|
||||
import hashlib
|
||||
import json
|
||||
from typing import Any
|
||||
import uuid
|
||||
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from govoplan_core.core.dataflows import (
|
||||
DataflowDatasetRequest,
|
||||
dataflow_dataset_output,
|
||||
)
|
||||
from govoplan_core.core.events import (
|
||||
EventActorRef,
|
||||
EventObjectRef,
|
||||
EventTenantRef,
|
||||
PlatformEvent,
|
||||
emit_platform_event,
|
||||
)
|
||||
from govoplan_core.security.module_permissions import scopes_grant_compatible
|
||||
from govoplan_core.security.time import utc_now
|
||||
from govoplan_reporting.backend.contracts import (
|
||||
CAPABILITY_REPORTING_CHART_RENDERER,
|
||||
ReportingChartRenderer,
|
||||
ReportingDatasetReadRequest,
|
||||
ReportingDatasetReadResult,
|
||||
ReportingReadModelProvider,
|
||||
ReportingRowPolicyProvider,
|
||||
ReportingRowPolicyRequest,
|
||||
capability,
|
||||
)
|
||||
from govoplan_reporting.backend.db.models import (
|
||||
ReportingExecution,
|
||||
ReportingQualityResult,
|
||||
)
|
||||
from govoplan_reporting.backend.definitions import get_definition, list_definitions
|
||||
from govoplan_reporting.backend.query_engine import (
|
||||
QUERY_ENGINE_VERSION,
|
||||
DefaultChartRenderer,
|
||||
QueryResult,
|
||||
execute_semantic_query,
|
||||
)
|
||||
from govoplan_reporting.backend.schemas import (
|
||||
DatasetDefinition,
|
||||
QualityPlanDefinition,
|
||||
ReportDefinition,
|
||||
ReportQuery,
|
||||
SemanticModelDefinition,
|
||||
)
|
||||
|
||||
|
||||
RUN_SCOPE = "reporting:report:run"
|
||||
QUALITY_SCOPE = "reporting:quality:run"
|
||||
|
||||
|
||||
class ReportingExecutionError(ValueError):
|
||||
pass
|
||||
|
||||
|
||||
class ReportingExecutionFailure(ReportingExecutionError):
|
||||
def __init__(self, message: str, execution_id: str) -> None:
|
||||
super().__init__(message)
|
||||
self.execution_id = execution_id
|
||||
|
||||
|
||||
class SqlReportingRunner:
|
||||
def __init__(self, registry: object | None) -> None:
|
||||
self.registry = registry
|
||||
|
||||
def execute(
|
||||
self,
|
||||
session: object,
|
||||
principal: object,
|
||||
*,
|
||||
report_id: str,
|
||||
report_revision: int | None = None,
|
||||
parameters: Mapping[str, object] | None = None,
|
||||
query: ReportQuery | None = None,
|
||||
idempotency_key: str,
|
||||
) -> Mapping[str, object]:
|
||||
return execute_report(
|
||||
_session(session),
|
||||
principal,
|
||||
registry=self.registry,
|
||||
report_id=report_id,
|
||||
report_revision=report_revision,
|
||||
parameters=parameters or {},
|
||||
query=query,
|
||||
idempotency_key=idempotency_key,
|
||||
)
|
||||
|
||||
def get_execution(
|
||||
self,
|
||||
session: object,
|
||||
principal: object,
|
||||
*,
|
||||
execution_id: str,
|
||||
) -> Mapping[str, object] | None:
|
||||
return get_execution(_session(session), principal, execution_id=execution_id)
|
||||
|
||||
|
||||
def execute_report(
|
||||
session: Session,
|
||||
principal: object,
|
||||
*,
|
||||
registry: object | None,
|
||||
report_id: str,
|
||||
report_revision: int | None,
|
||||
parameters: Mapping[str, object],
|
||||
query: ReportQuery | None,
|
||||
idempotency_key: str,
|
||||
) -> dict[str, object]:
|
||||
_require_scope(principal, RUN_SCOPE)
|
||||
report_record = get_definition(
|
||||
session,
|
||||
principal,
|
||||
definition_kind="report",
|
||||
definition_id=report_id,
|
||||
revision=report_revision,
|
||||
)
|
||||
if report_record is None:
|
||||
raise LookupError("Reporting report definition not found.")
|
||||
if report_record.status != "active":
|
||||
raise ReportingExecutionError("Only active report definitions can run.")
|
||||
report = ReportDefinition.model_validate(report_record.payload)
|
||||
semantic_record = get_definition(
|
||||
session,
|
||||
principal,
|
||||
definition_kind="semantic_model",
|
||||
definition_id=report.semantic_model_id,
|
||||
revision=report.semantic_model_revision,
|
||||
)
|
||||
if semantic_record is None or semantic_record.status != "active":
|
||||
raise ReportingExecutionError(
|
||||
"The report's pinned semantic model is unavailable or inactive."
|
||||
)
|
||||
semantic = SemanticModelDefinition.model_validate(semantic_record.payload)
|
||||
dataset_record = get_definition(
|
||||
session,
|
||||
principal,
|
||||
definition_kind="dataset",
|
||||
definition_id=semantic.dataset_id,
|
||||
revision=semantic.dataset_revision,
|
||||
)
|
||||
if dataset_record is None or dataset_record.status != "active":
|
||||
raise ReportingExecutionError(
|
||||
"The report's pinned analytical dataset is unavailable or inactive."
|
||||
)
|
||||
dataset = DatasetDefinition.model_validate(dataset_record.payload)
|
||||
bound_parameters = _bind_parameters(report, parameters)
|
||||
effective_query = query or report.default_query
|
||||
clean_idempotency_key = _required(
|
||||
idempotency_key,
|
||||
"Reporting execution idempotency key",
|
||||
255,
|
||||
)
|
||||
request_sha256 = _sha256(
|
||||
{
|
||||
"report_id": report_id,
|
||||
"report_revision": report_record.revision,
|
||||
"semantic_model_id": semantic_record.definition_id,
|
||||
"semantic_model_revision": semantic_record.revision,
|
||||
"dataset_id": dataset_record.definition_id,
|
||||
"dataset_revision": dataset_record.revision,
|
||||
"parameters": bound_parameters,
|
||||
"query": effective_query.model_dump(mode="json"),
|
||||
}
|
||||
)
|
||||
replay = _execution_replay(
|
||||
session,
|
||||
tenant_id=_tenant(principal),
|
||||
idempotency_key=clean_idempotency_key,
|
||||
request_sha256=request_sha256,
|
||||
)
|
||||
if replay is not None:
|
||||
return _execution_payload(replay, report=report, registry=registry)
|
||||
started_at = utc_now()
|
||||
execution = ReportingExecution(
|
||||
tenant_id=_tenant(principal),
|
||||
execution_id=str(uuid.uuid4()),
|
||||
report_id=report_record.definition_id,
|
||||
report_revision=report_record.revision,
|
||||
semantic_model_id=semantic_record.definition_id,
|
||||
semantic_model_revision=semantic_record.revision,
|
||||
dataset_id=dataset_record.definition_id,
|
||||
dataset_revision=dataset_record.revision,
|
||||
status="running",
|
||||
idempotency_key=clean_idempotency_key,
|
||||
request_sha256=request_sha256,
|
||||
parameters=_json_value(bound_parameters),
|
||||
query=effective_query.model_dump(mode="json"),
|
||||
definition_hashes={
|
||||
"report": report_record.content_hash,
|
||||
"semantic_model": semantic_record.content_hash,
|
||||
"dataset": dataset_record.content_hash,
|
||||
},
|
||||
started_at=started_at,
|
||||
actor_id=_actor(principal),
|
||||
)
|
||||
session.add(execution)
|
||||
session.flush()
|
||||
try:
|
||||
source = _read_dataset(
|
||||
session,
|
||||
principal,
|
||||
registry=registry,
|
||||
dataset=dataset,
|
||||
parameters=bound_parameters,
|
||||
)
|
||||
diagnostics = list(source.diagnostics)
|
||||
diagnostics.extend(_freshness_diagnostics(dataset, source, now=started_at))
|
||||
normalized_rows = tuple(_json_value(dict(row)) for row in source.rows)
|
||||
diagnostics.extend(_validate_schema(dataset, normalized_rows))
|
||||
authorized_rows, policy_provenance = _apply_row_policy(
|
||||
session,
|
||||
principal,
|
||||
registry=registry,
|
||||
dataset_id=dataset_record.definition_id,
|
||||
dataset_revision=dataset_record.revision,
|
||||
dataset=dataset,
|
||||
rows=normalized_rows,
|
||||
)
|
||||
quality_results = _evaluate_blocking_quality_plans(
|
||||
session,
|
||||
principal,
|
||||
dataset_id=dataset_record.definition_id,
|
||||
dataset_revision=dataset_record.revision,
|
||||
rows=authorized_rows,
|
||||
output_hash=source.output_hash,
|
||||
source_fingerprints=source.source_fingerprints,
|
||||
)
|
||||
result = execute_semantic_query(authorized_rows, semantic, effective_query)
|
||||
output_hash = _sha256(
|
||||
{
|
||||
"rows": result.rows,
|
||||
"schema": result.schema,
|
||||
"query": effective_query.model_dump(mode="json"),
|
||||
"source_output_hash": source.output_hash,
|
||||
}
|
||||
)
|
||||
execution.status = "succeeded"
|
||||
execution.source_fingerprints = _json_value(source.source_fingerprints)
|
||||
execution.output_hash = output_hash
|
||||
execution.executor_version = f"{QUERY_ENGINE_VERSION}+{source.executor_version}"
|
||||
execution.result_schema = list(result.schema)
|
||||
execution.result_rows = list(result.rows)
|
||||
execution.total_rows = result.total_rows
|
||||
execution.truncated = result.truncated or source.truncated
|
||||
execution.diagnostics = _json_value(diagnostics)
|
||||
execution.provenance = {
|
||||
"source": _json_value(source.provenance),
|
||||
"source_output_hash": source.output_hash,
|
||||
"row_policy": _json_value(policy_provenance),
|
||||
"quality_result_ids": [item.result_id for item in quality_results],
|
||||
"authorized_source_rows": len(authorized_rows),
|
||||
"report_content_hash": report_record.content_hash,
|
||||
"semantic_model_content_hash": semantic_record.content_hash,
|
||||
"dataset_content_hash": dataset_record.content_hash,
|
||||
}
|
||||
execution.finished_at = utc_now()
|
||||
session.flush()
|
||||
_emit_execution_event(session, execution, report_record.name)
|
||||
return _execution_payload(execution, report=report, registry=registry)
|
||||
except Exception as exc:
|
||||
execution.status = "failed"
|
||||
execution.finished_at = utc_now()
|
||||
execution.diagnostics = [
|
||||
{
|
||||
"severity": "error",
|
||||
"code": "report_execution_failed",
|
||||
"message": str(exc),
|
||||
}
|
||||
]
|
||||
session.flush()
|
||||
_emit_execution_event(session, execution, report_record.name)
|
||||
raise ReportingExecutionFailure(str(exc), execution.execution_id) from exc
|
||||
|
||||
|
||||
def get_execution(
|
||||
session: Session,
|
||||
principal: object,
|
||||
*,
|
||||
execution_id: str,
|
||||
) -> dict[str, object] | None:
|
||||
row = (
|
||||
session.query(ReportingExecution)
|
||||
.filter(
|
||||
ReportingExecution.tenant_id == _tenant(principal),
|
||||
ReportingExecution.execution_id == execution_id,
|
||||
)
|
||||
.one_or_none()
|
||||
)
|
||||
if row is None:
|
||||
return None
|
||||
report_record = get_definition(
|
||||
session,
|
||||
principal,
|
||||
definition_kind="report",
|
||||
definition_id=row.report_id,
|
||||
revision=row.report_revision,
|
||||
)
|
||||
if report_record is None:
|
||||
return None
|
||||
return _execution_payload(
|
||||
row,
|
||||
report=ReportDefinition.model_validate(report_record.payload),
|
||||
registry=None,
|
||||
)
|
||||
|
||||
|
||||
def list_executions(
|
||||
session: Session,
|
||||
principal: object,
|
||||
*,
|
||||
report_id: str,
|
||||
limit: int = 100,
|
||||
) -> tuple[dict[str, object], ...]:
|
||||
if (
|
||||
get_definition(
|
||||
session,
|
||||
principal,
|
||||
definition_kind="report",
|
||||
definition_id=report_id,
|
||||
)
|
||||
is None
|
||||
):
|
||||
return ()
|
||||
rows = (
|
||||
session.query(ReportingExecution)
|
||||
.filter(
|
||||
ReportingExecution.tenant_id == _tenant(principal),
|
||||
ReportingExecution.report_id == report_id,
|
||||
)
|
||||
.order_by(ReportingExecution.started_at.desc())
|
||||
.limit(max(1, min(limit, 200)))
|
||||
.all()
|
||||
)
|
||||
return tuple(_execution_payload(row, report=None, registry=None) for row in rows)
|
||||
|
||||
|
||||
def _read_dataset(
|
||||
session: Session,
|
||||
principal: object,
|
||||
*,
|
||||
registry: object | None,
|
||||
dataset: DatasetDefinition,
|
||||
parameters: Mapping[str, object],
|
||||
) -> ReportingDatasetReadResult:
|
||||
source_parameters = {**dataset.source_parameters, **parameters}
|
||||
if dataset.source_kind == "static":
|
||||
rows = tuple(dict(item) for item in dataset.static_rows)
|
||||
return ReportingDatasetReadResult(
|
||||
rows=rows,
|
||||
total_rows=len(rows),
|
||||
truncated=False,
|
||||
output_hash=_sha256(rows),
|
||||
executor_version="reporting-static-v1",
|
||||
definition_hash=dataset.definition_hash,
|
||||
generated_at=utc_now(),
|
||||
provenance={"source_kind": "static", "source_ref": dataset.source_ref},
|
||||
)
|
||||
if dataset.source_kind == "dataflow":
|
||||
provider = dataflow_dataset_output(registry)
|
||||
if provider is None:
|
||||
raise ReportingExecutionError(
|
||||
"The Dataflow dataset provider is not enabled."
|
||||
)
|
||||
result = provider.read_output(
|
||||
session,
|
||||
principal,
|
||||
request=DataflowDatasetRequest(
|
||||
pipeline_ref=dataset.source_ref,
|
||||
revision=dataset.source_revision or 0,
|
||||
parameters=source_parameters,
|
||||
row_limit=2_000,
|
||||
expected_definition_hash=dataset.definition_hash,
|
||||
expected_source_fingerprints=tuple(
|
||||
dataset.expected_source_fingerprints
|
||||
),
|
||||
),
|
||||
)
|
||||
return ReportingDatasetReadResult(
|
||||
rows=result.rows,
|
||||
total_rows=result.total_rows,
|
||||
truncated=result.truncated,
|
||||
output_hash=result.output_hash,
|
||||
executor_version=result.executor_version,
|
||||
definition_hash=result.definition_hash,
|
||||
source_fingerprints=result.source_fingerprints,
|
||||
diagnostics=result.diagnostics,
|
||||
generated_at=result.generated_at,
|
||||
provenance=result.provenance,
|
||||
)
|
||||
provider = capability(registry, dataset.source_ref)
|
||||
if not isinstance(provider, ReportingReadModelProvider):
|
||||
raise ReportingExecutionError(
|
||||
f"Reporting read-model provider {dataset.source_ref!r} is unavailable."
|
||||
)
|
||||
return provider.read_dataset(
|
||||
session,
|
||||
principal,
|
||||
request=ReportingDatasetReadRequest(
|
||||
source_ref=dataset.source_ref,
|
||||
source_revision=dataset.source_revision,
|
||||
parameters=source_parameters,
|
||||
expected_definition_hash=dataset.definition_hash,
|
||||
expected_source_fingerprints=tuple(dataset.expected_source_fingerprints),
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def _apply_row_policy(
|
||||
session: Session,
|
||||
principal: object,
|
||||
*,
|
||||
registry: object | None,
|
||||
dataset_id: str,
|
||||
dataset_revision: int,
|
||||
dataset: DatasetDefinition,
|
||||
rows: tuple[Mapping[str, object], ...],
|
||||
) -> tuple[tuple[Mapping[str, object], ...], Mapping[str, object]]:
|
||||
if dataset.row_policy_ref is None:
|
||||
return rows, {"mode": "dataset_access_only"}
|
||||
provider = capability(registry, dataset.row_policy_ref)
|
||||
if not isinstance(provider, ReportingRowPolicyProvider):
|
||||
raise ReportingExecutionError(
|
||||
f"Required row-policy provider {dataset.row_policy_ref!r} is unavailable."
|
||||
)
|
||||
result = provider.authorize_rows(
|
||||
session,
|
||||
principal,
|
||||
request=ReportingRowPolicyRequest(
|
||||
dataset_id=dataset_id,
|
||||
dataset_revision=dataset_revision,
|
||||
policy_ref=dataset.row_policy_ref,
|
||||
rows=rows,
|
||||
),
|
||||
)
|
||||
if len(result.rows) > len(rows):
|
||||
raise ReportingExecutionError(
|
||||
"A row-policy provider returned more rows than it received."
|
||||
)
|
||||
return result.rows, {
|
||||
"mode": "provider",
|
||||
"provider": dataset.row_policy_ref,
|
||||
"decision_ref": result.decision_ref,
|
||||
**dict(result.provenance),
|
||||
}
|
||||
|
||||
|
||||
def _freshness_diagnostics(
|
||||
dataset: DatasetDefinition,
|
||||
source: ReportingDatasetReadResult,
|
||||
*,
|
||||
now: datetime,
|
||||
) -> tuple[dict[str, object], ...]:
|
||||
policy = dataset.freshness
|
||||
if (
|
||||
policy.require_source_fingerprints
|
||||
and not source.source_fingerprints
|
||||
and dataset.source_kind != "static"
|
||||
):
|
||||
raise ReportingExecutionError(
|
||||
"The analytical dataset requires source fingerprints, but the provider returned none."
|
||||
)
|
||||
if policy.max_age_seconds is None:
|
||||
return ()
|
||||
generated_at = source.generated_at
|
||||
if generated_at is None:
|
||||
message = "The dataset provider did not report a generation timestamp."
|
||||
if policy.stale_action == "block":
|
||||
raise ReportingExecutionError(message)
|
||||
return (
|
||||
{"severity": "warning", "code": "freshness_unknown", "message": message},
|
||||
)
|
||||
if generated_at.tzinfo is None:
|
||||
generated_at = generated_at.replace(tzinfo=UTC)
|
||||
age_seconds = max(0, int((now - generated_at).total_seconds()))
|
||||
if age_seconds <= policy.max_age_seconds:
|
||||
return ()
|
||||
message = (
|
||||
f"Dataset age {age_seconds}s exceeds the configured "
|
||||
f"{policy.max_age_seconds}s freshness limit."
|
||||
)
|
||||
if policy.stale_action == "block":
|
||||
raise ReportingExecutionError(message)
|
||||
if policy.stale_action == "warn":
|
||||
return ({"severity": "warning", "code": "dataset_stale", "message": message},)
|
||||
return ()
|
||||
|
||||
|
||||
def _validate_schema(
|
||||
dataset: DatasetDefinition,
|
||||
rows: Sequence[Mapping[str, object]],
|
||||
) -> tuple[dict[str, object], ...]:
|
||||
if not dataset.fields:
|
||||
return (
|
||||
{
|
||||
"severity": "info",
|
||||
"code": "schema_inferred",
|
||||
"message": "Dataset schema is inferred because no explicit fields are pinned.",
|
||||
},
|
||||
)
|
||||
errors: list[str] = []
|
||||
for field in dataset.fields:
|
||||
for index, row in enumerate(rows):
|
||||
value = row.get(field.name)
|
||||
if value is None:
|
||||
if not field.nullable:
|
||||
errors.append(f"row {index + 1}: {field.name} is null")
|
||||
continue
|
||||
if not _matches_type(value, field.type):
|
||||
errors.append(
|
||||
f"row {index + 1}: {field.name} does not match {field.type}"
|
||||
)
|
||||
if len(errors) >= 20:
|
||||
break
|
||||
if len(errors) >= 20:
|
||||
break
|
||||
if errors:
|
||||
raise ReportingExecutionError(
|
||||
"Analytical dataset schema validation failed: " + "; ".join(errors)
|
||||
)
|
||||
return ()
|
||||
|
||||
|
||||
def _evaluate_blocking_quality_plans(
|
||||
session: Session,
|
||||
principal: object,
|
||||
*,
|
||||
dataset_id: str,
|
||||
dataset_revision: int,
|
||||
rows: tuple[Mapping[str, object], ...],
|
||||
output_hash: str,
|
||||
source_fingerprints: Sequence[Mapping[str, object]],
|
||||
) -> tuple[ReportingQualityResult, ...]:
|
||||
plans, _total = list_definitions(
|
||||
session,
|
||||
principal,
|
||||
definition_kinds=("quality_plan",),
|
||||
statuses=("active",),
|
||||
limit=200,
|
||||
)
|
||||
relevant = tuple(
|
||||
plan
|
||||
for plan in plans
|
||||
if plan.parent_id == dataset_id and plan.parent_revision == dataset_revision
|
||||
)
|
||||
results: list[ReportingQualityResult] = []
|
||||
for plan_record in relevant:
|
||||
plan = QualityPlanDefinition.model_validate(plan_record.payload)
|
||||
result = _record_quality_result(
|
||||
session,
|
||||
principal,
|
||||
plan_id=plan_record.definition_id,
|
||||
plan_revision=plan_record.revision,
|
||||
dataset_id=dataset_id,
|
||||
dataset_revision=dataset_revision,
|
||||
plan=plan,
|
||||
rows=rows,
|
||||
output_hash=output_hash,
|
||||
source_fingerprints=source_fingerprints,
|
||||
)
|
||||
results.append(result)
|
||||
if plan.block_report_execution and result.status == "failed":
|
||||
raise ReportingExecutionError(
|
||||
f"Quality plan {plan_record.name!r} blocked report execution."
|
||||
)
|
||||
return tuple(results)
|
||||
|
||||
|
||||
def run_quality_plan(
|
||||
session: Session,
|
||||
principal: object,
|
||||
*,
|
||||
registry: object | None,
|
||||
quality_plan_id: str,
|
||||
quality_plan_revision: int | None,
|
||||
parameters: Mapping[str, object],
|
||||
) -> dict[str, object]:
|
||||
_require_scope(principal, QUALITY_SCOPE)
|
||||
record = get_definition(
|
||||
session,
|
||||
principal,
|
||||
definition_kind="quality_plan",
|
||||
definition_id=quality_plan_id,
|
||||
revision=quality_plan_revision,
|
||||
)
|
||||
if record is None or record.status != "active":
|
||||
raise LookupError("Active Reporting quality plan not found.")
|
||||
plan = QualityPlanDefinition.model_validate(record.payload)
|
||||
dataset_record = get_definition(
|
||||
session,
|
||||
principal,
|
||||
definition_kind="dataset",
|
||||
definition_id=plan.dataset_id,
|
||||
revision=plan.dataset_revision,
|
||||
)
|
||||
if dataset_record is None or dataset_record.status != "active":
|
||||
raise ReportingExecutionError("Quality plan dataset is unavailable.")
|
||||
dataset = DatasetDefinition.model_validate(dataset_record.payload)
|
||||
source = _read_dataset(
|
||||
session,
|
||||
principal,
|
||||
registry=registry,
|
||||
dataset=dataset,
|
||||
parameters=parameters,
|
||||
)
|
||||
rows = tuple(_json_value(dict(item)) for item in source.rows)
|
||||
result = _record_quality_result(
|
||||
session,
|
||||
principal,
|
||||
plan_id=record.definition_id,
|
||||
plan_revision=record.revision,
|
||||
dataset_id=dataset_record.definition_id,
|
||||
dataset_revision=dataset_record.revision,
|
||||
plan=plan,
|
||||
rows=rows,
|
||||
output_hash=source.output_hash,
|
||||
source_fingerprints=source.source_fingerprints,
|
||||
)
|
||||
return _quality_payload(result)
|
||||
|
||||
|
||||
def _record_quality_result(
|
||||
session: Session,
|
||||
principal: object,
|
||||
*,
|
||||
plan_id: str,
|
||||
plan_revision: int,
|
||||
dataset_id: str,
|
||||
dataset_revision: int,
|
||||
plan: QualityPlanDefinition,
|
||||
rows: Sequence[Mapping[str, object]],
|
||||
output_hash: str,
|
||||
source_fingerprints: Sequence[Mapping[str, object]],
|
||||
) -> ReportingQualityResult:
|
||||
assertions = [_evaluate_assertion(item, rows) for item in plan.assertions]
|
||||
failed = any(
|
||||
not item["passed"] and item["severity"] in {"error", "blocker"}
|
||||
for item in assertions
|
||||
)
|
||||
warning = any(not item["passed"] for item in assertions)
|
||||
row = ReportingQualityResult(
|
||||
tenant_id=_tenant(principal),
|
||||
result_id=str(uuid.uuid4()),
|
||||
quality_plan_id=plan_id,
|
||||
quality_plan_revision=plan_revision,
|
||||
dataset_id=dataset_id,
|
||||
dataset_revision=dataset_revision,
|
||||
status="failed" if failed else "warning" if warning else "passed",
|
||||
output_hash=output_hash,
|
||||
assertions=assertions,
|
||||
source_fingerprints=_json_value(source_fingerprints),
|
||||
evaluated_at=utc_now(),
|
||||
actor_id=_actor(principal),
|
||||
)
|
||||
session.add(row)
|
||||
session.flush()
|
||||
return row
|
||||
|
||||
|
||||
def _evaluate_assertion(
|
||||
assertion, rows: Sequence[Mapping[str, object]]
|
||||
) -> dict[str, object]:
|
||||
field = assertion.field
|
||||
config = assertion.config
|
||||
failures = 0
|
||||
if assertion.kind == "not_null":
|
||||
failures = sum(item.get(field or "") is None for item in rows)
|
||||
elif assertion.kind == "unique":
|
||||
values = [item.get(field or "") for item in rows]
|
||||
failures = len(values) - len({_stable_value(item) for item in values})
|
||||
elif assertion.kind == "range":
|
||||
minimum = config.get("minimum")
|
||||
maximum = config.get("maximum")
|
||||
failures = sum(
|
||||
value is not None
|
||||
and (
|
||||
(minimum is not None and value < minimum)
|
||||
or (maximum is not None and value > maximum)
|
||||
)
|
||||
for value in (item.get(field or "") for item in rows)
|
||||
)
|
||||
elif assertion.kind == "accepted_values":
|
||||
accepted = {_stable_value(item) for item in config.get("values", [])}
|
||||
failures = sum(
|
||||
_stable_value(item.get(field or "")) not in accepted for item in rows
|
||||
)
|
||||
elif assertion.kind == "row_count":
|
||||
minimum = int(config.get("minimum", 0))
|
||||
maximum = int(config.get("maximum", 2_000_000_000))
|
||||
failures = 0 if minimum <= len(rows) <= maximum else 1
|
||||
elif assertion.kind == "comparison":
|
||||
other_field = str(config.get("other_field") or "")
|
||||
operator = str(config.get("operator") or "eq")
|
||||
if not field or not other_field or operator not in {"eq", "ne"}:
|
||||
raise ReportingExecutionError("Comparison quality assertion is invalid.")
|
||||
failures = sum(
|
||||
(item.get(field) == item.get(other_field)) != (operator == "eq")
|
||||
for item in rows
|
||||
)
|
||||
else:
|
||||
raise ReportingExecutionError(
|
||||
f"Unsupported quality assertion: {assertion.kind!r}."
|
||||
)
|
||||
return {
|
||||
"key": assertion.key,
|
||||
"kind": assertion.kind,
|
||||
"severity": assertion.severity,
|
||||
"passed": failures == 0,
|
||||
"failure_count": failures,
|
||||
"row_count": len(rows),
|
||||
}
|
||||
|
||||
|
||||
def _execution_payload(
|
||||
row: ReportingExecution,
|
||||
*,
|
||||
report: ReportDefinition | None,
|
||||
registry: object | None,
|
||||
) -> dict[str, object]:
|
||||
payload: dict[str, object] = {
|
||||
"execution_id": row.execution_id,
|
||||
"report_id": row.report_id,
|
||||
"report_revision": row.report_revision,
|
||||
"semantic_model_id": row.semantic_model_id,
|
||||
"semantic_model_revision": row.semantic_model_revision,
|
||||
"dataset_id": row.dataset_id,
|
||||
"dataset_revision": row.dataset_revision,
|
||||
"status": row.status,
|
||||
"parameters": dict(row.parameters or {}),
|
||||
"query": dict(row.query or {}),
|
||||
"definition_hashes": dict(row.definition_hashes or {}),
|
||||
"source_fingerprints": list(row.source_fingerprints or []),
|
||||
"output_hash": row.output_hash,
|
||||
"executor_version": row.executor_version,
|
||||
"schema": list(row.result_schema or []),
|
||||
"rows": list(row.result_rows or []),
|
||||
"total_rows": row.total_rows,
|
||||
"truncated": row.truncated,
|
||||
"diagnostics": list(row.diagnostics or []),
|
||||
"provenance": dict(row.provenance or {}),
|
||||
"started_at": _datetime_text(row.started_at),
|
||||
"finished_at": _datetime_text(row.finished_at),
|
||||
"actor_id": row.actor_id,
|
||||
}
|
||||
if row.status == "succeeded" and report is not None:
|
||||
result = QueryResult(
|
||||
rows=tuple(dict(item) for item in row.result_rows or []),
|
||||
total_rows=row.total_rows,
|
||||
schema=tuple(dict(item) for item in row.result_schema or []),
|
||||
truncated=row.truncated,
|
||||
)
|
||||
renderer = capability(registry, CAPABILITY_REPORTING_CHART_RENDERER)
|
||||
if not isinstance(renderer, ReportingChartRenderer):
|
||||
renderer = DefaultChartRenderer()
|
||||
payload["visualization"] = dict(
|
||||
renderer.render(visualization=report.visualization, result=result)
|
||||
)
|
||||
return payload
|
||||
|
||||
|
||||
def _quality_payload(row: ReportingQualityResult) -> dict[str, object]:
|
||||
return {
|
||||
"result_id": row.result_id,
|
||||
"quality_plan_id": row.quality_plan_id,
|
||||
"quality_plan_revision": row.quality_plan_revision,
|
||||
"dataset_id": row.dataset_id,
|
||||
"dataset_revision": row.dataset_revision,
|
||||
"status": row.status,
|
||||
"output_hash": row.output_hash,
|
||||
"assertions": list(row.assertions or []),
|
||||
"source_fingerprints": list(row.source_fingerprints or []),
|
||||
"evaluated_at": _datetime_text(row.evaluated_at),
|
||||
}
|
||||
|
||||
|
||||
def _execution_replay(
|
||||
session: Session,
|
||||
*,
|
||||
tenant_id: str,
|
||||
idempotency_key: str,
|
||||
request_sha256: str,
|
||||
) -> ReportingExecution | None:
|
||||
row = (
|
||||
session.query(ReportingExecution)
|
||||
.filter(
|
||||
ReportingExecution.tenant_id == tenant_id,
|
||||
ReportingExecution.idempotency_key == idempotency_key,
|
||||
)
|
||||
.one_or_none()
|
||||
)
|
||||
if row is not None and row.request_sha256 != request_sha256:
|
||||
raise ReportingExecutionError(
|
||||
"Reporting execution idempotency conflict: the key belongs to another request."
|
||||
)
|
||||
return row
|
||||
|
||||
|
||||
def _bind_parameters(
|
||||
report: ReportDefinition,
|
||||
supplied: Mapping[str, object],
|
||||
) -> dict[str, object]:
|
||||
definitions = {item.key: item for item in report.parameters}
|
||||
unknown = set(supplied) - set(definitions)
|
||||
if unknown:
|
||||
raise ReportingExecutionError(
|
||||
"Unknown report parameters: " + ", ".join(sorted(unknown))
|
||||
)
|
||||
result: dict[str, object] = {}
|
||||
for key, definition in definitions.items():
|
||||
value = supplied.get(key, definition.default)
|
||||
if value is None and definition.required:
|
||||
raise ReportingExecutionError(f"Report parameter {key!r} is required.")
|
||||
if definition.allowed_values and value not in definition.allowed_values:
|
||||
raise ReportingExecutionError(
|
||||
f"Report parameter {key!r} is outside its allowed values."
|
||||
)
|
||||
if value is not None:
|
||||
result[key] = value
|
||||
return result
|
||||
|
||||
|
||||
def _emit_execution_event(
|
||||
session: Session,
|
||||
execution: ReportingExecution,
|
||||
report_name: str,
|
||||
) -> None:
|
||||
emit_platform_event(
|
||||
session,
|
||||
PlatformEvent(
|
||||
type=f"reporting.execution.{execution.status}",
|
||||
module_id="reporting",
|
||||
payload={
|
||||
"execution_id": execution.execution_id,
|
||||
"report_id": execution.report_id,
|
||||
"report_revision": execution.report_revision,
|
||||
"output_hash": execution.output_hash,
|
||||
"total_rows": execution.total_rows,
|
||||
"truncated": execution.truncated,
|
||||
},
|
||||
occurred_at=execution.finished_at or execution.started_at,
|
||||
actor=EventActorRef(type="account", id=execution.actor_id),
|
||||
tenant=EventTenantRef(id=execution.tenant_id),
|
||||
resource=EventObjectRef(
|
||||
type="report_execution",
|
||||
id=execution.execution_id,
|
||||
label=report_name,
|
||||
),
|
||||
classification="internal",
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def _matches_type(value: object, field_type: str) -> bool:
|
||||
if field_type == "string":
|
||||
return isinstance(value, str)
|
||||
if field_type == "integer":
|
||||
return isinstance(value, int) and not isinstance(value, bool)
|
||||
if field_type == "number":
|
||||
return isinstance(value, (int, float)) and not isinstance(value, bool)
|
||||
if field_type == "boolean":
|
||||
return isinstance(value, bool)
|
||||
if field_type in {"date", "datetime"}:
|
||||
if isinstance(value, (date, datetime)):
|
||||
return True
|
||||
if isinstance(value, str):
|
||||
try:
|
||||
datetime.fromisoformat(value.replace("Z", "+00:00"))
|
||||
return True
|
||||
except ValueError:
|
||||
return False
|
||||
return False
|
||||
if field_type == "json":
|
||||
return isinstance(value, (dict, list, tuple))
|
||||
return False
|
||||
|
||||
|
||||
def _stable_value(value: object) -> str:
|
||||
return json.dumps(_json_value(value), sort_keys=True, separators=(",", ":"))
|
||||
|
||||
|
||||
def _required(value: object, label: str, maximum: int) -> str:
|
||||
result = str(value or "").strip()
|
||||
if not result:
|
||||
raise ReportingExecutionError(f"{label} is required.")
|
||||
if len(result) > maximum:
|
||||
raise ReportingExecutionError(f"{label} is limited to {maximum} characters.")
|
||||
return result
|
||||
|
||||
|
||||
def _require_scope(principal: object, scope: str) -> None:
|
||||
method = getattr(principal, "has", None)
|
||||
allowed = (
|
||||
bool(method(scope))
|
||||
if callable(method)
|
||||
else scopes_grant_compatible(
|
||||
frozenset(getattr(principal, "scopes", ()) or ()), scope
|
||||
)
|
||||
)
|
||||
if not allowed:
|
||||
raise PermissionError(f"Missing scope: {scope}")
|
||||
|
||||
|
||||
def _tenant(principal: object) -> str:
|
||||
tenant_id = str(getattr(principal, "tenant_id", "") or "").strip()
|
||||
if not tenant_id:
|
||||
raise ReportingExecutionError(
|
||||
"Reporting execution requires a tenant-bound principal."
|
||||
)
|
||||
return tenant_id
|
||||
|
||||
|
||||
def _actor(principal: object) -> str | None:
|
||||
for value in (
|
||||
getattr(principal, "account_id", None),
|
||||
getattr(principal, "identity_id", None),
|
||||
getattr(principal, "membership_id", None),
|
||||
):
|
||||
if str(value or "").strip():
|
||||
return str(value)
|
||||
return None
|
||||
|
||||
|
||||
def _session(value: object) -> Session:
|
||||
if not isinstance(value, Session):
|
||||
raise TypeError("Reporting execution requires a SQLAlchemy session.")
|
||||
return value
|
||||
|
||||
|
||||
def _sha256(value: object) -> str:
|
||||
return hashlib.sha256(
|
||||
json.dumps(
|
||||
_json_value(value),
|
||||
sort_keys=True,
|
||||
separators=(",", ":"),
|
||||
ensure_ascii=True,
|
||||
).encode("utf-8")
|
||||
).hexdigest()
|
||||
|
||||
|
||||
def _json_value(value: object) -> Any:
|
||||
if isinstance(value, (date, datetime)):
|
||||
return value.isoformat()
|
||||
if hasattr(value, "model_dump"):
|
||||
return _json_value(value.model_dump(mode="json"))
|
||||
if isinstance(value, Mapping):
|
||||
return {str(key): _json_value(item) for key, item in value.items()}
|
||||
if isinstance(value, (list, tuple)):
|
||||
return [_json_value(item) for item in value]
|
||||
return value
|
||||
|
||||
|
||||
def _datetime_text(value: datetime | None) -> str | None:
|
||||
if value is None:
|
||||
return None
|
||||
if value.tzinfo is None:
|
||||
value = value.replace(tzinfo=UTC)
|
||||
return value.isoformat()
|
||||
|
||||
|
||||
__all__ = [
|
||||
"QUALITY_SCOPE",
|
||||
"RUN_SCOPE",
|
||||
"ReportingExecutionError",
|
||||
"ReportingExecutionFailure",
|
||||
"SqlReportingRunner",
|
||||
"execute_report",
|
||||
"get_execution",
|
||||
"list_executions",
|
||||
"run_quality_plan",
|
||||
]
|
||||
Reference in New Issue
Block a user