1239 lines
41 KiB
Python
1239 lines
41 KiB
Python
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.domain import ReportingDefinitionRecord
|
|
from govoplan_reporting.backend.governance import require_definition_action
|
|
from govoplan_reporting.backend.postgres_planner import (
|
|
POSTGRES_PLANNER_VERSION,
|
|
execute_postgres_query,
|
|
)
|
|
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,
|
|
registry=self.registry,
|
|
)
|
|
|
|
|
|
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)
|
|
report_decision = require_definition_action(
|
|
session,
|
|
principal,
|
|
registry=registry,
|
|
record=report_record,
|
|
action="run",
|
|
)
|
|
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)
|
|
semantic_decision = require_definition_action(
|
|
session,
|
|
principal,
|
|
registry=registry,
|
|
record=semantic_record,
|
|
action="view",
|
|
)
|
|
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)
|
|
dataset_decision = require_definition_action(
|
|
session,
|
|
principal,
|
|
registry=registry,
|
|
record=dataset_record,
|
|
action="view",
|
|
)
|
|
bound_parameters = _bind_parameters(report, parameters)
|
|
effective_query = _enforce_query_access(report, 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:
|
|
delivery = _authorize_execution_delivery(
|
|
session,
|
|
principal,
|
|
registry=registry,
|
|
row=replay,
|
|
report_record=report_record,
|
|
)
|
|
return _execution_payload(
|
|
replay,
|
|
report=report,
|
|
registry=registry,
|
|
delivery_authorization=delivery,
|
|
)
|
|
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_postgres_query(
|
|
session,
|
|
rows=authorized_rows,
|
|
dataset=dataset,
|
|
semantic_model=semantic,
|
|
query=effective_query,
|
|
) or execute_semantic_query(authorized_rows, semantic, effective_query)
|
|
diagnostics.extend(result.diagnostics)
|
|
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
|
|
planner_version = (
|
|
POSTGRES_PLANNER_VERSION
|
|
if any(item.get("code") == "postgresql_semantic_plan" for item in result.diagnostics)
|
|
else QUERY_ENGINE_VERSION
|
|
)
|
|
execution.executor_version = f"{planner_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,
|
|
"definition_governance": {
|
|
"report": report_decision.to_dict(),
|
|
"semantic_model": semantic_decision.to_dict(),
|
|
"dataset": dataset_decision.to_dict(),
|
|
},
|
|
"access_explanation": _access_explanation(
|
|
report,
|
|
effective_query,
|
|
source_rows=len(normalized_rows),
|
|
authorized_rows=len(authorized_rows),
|
|
row_policy=policy_provenance,
|
|
),
|
|
}
|
|
execution.finished_at = utc_now()
|
|
session.flush()
|
|
_emit_execution_event(session, execution, report_record.name)
|
|
delivery = _authorize_execution_delivery(
|
|
session,
|
|
principal,
|
|
registry=registry,
|
|
row=execution,
|
|
report_record=report_record,
|
|
)
|
|
return _execution_payload(
|
|
execution,
|
|
report=report,
|
|
registry=registry,
|
|
delivery_authorization=delivery,
|
|
)
|
|
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,
|
|
registry: object | None = None,
|
|
) -> 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
|
|
delivery = _authorize_execution_delivery(
|
|
session,
|
|
principal,
|
|
registry=registry,
|
|
row=row,
|
|
report_record=report_record,
|
|
)
|
|
return _execution_payload(
|
|
row,
|
|
report=ReportDefinition.model_validate(report_record.payload),
|
|
registry=registry,
|
|
delivery_authorization=delivery,
|
|
)
|
|
|
|
|
|
def list_executions(
|
|
session: Session,
|
|
principal: object,
|
|
*,
|
|
report_id: str,
|
|
limit: int = 100,
|
|
registry: object | None = None,
|
|
) -> tuple[dict[str, object], ...]:
|
|
current_report = get_definition(
|
|
session,
|
|
principal,
|
|
definition_kind="report",
|
|
definition_id=report_id,
|
|
)
|
|
if current_report 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()
|
|
)
|
|
authorization_cache: dict[tuple[int, int, int], dict[str, object]] = {}
|
|
payloads: list[dict[str, object]] = []
|
|
for row in rows:
|
|
key = (
|
|
row.report_revision,
|
|
row.semantic_model_revision,
|
|
row.dataset_revision,
|
|
)
|
|
report_record = (
|
|
current_report
|
|
if current_report.revision == row.report_revision
|
|
else get_definition(
|
|
session,
|
|
principal,
|
|
definition_kind="report",
|
|
definition_id=row.report_id,
|
|
revision=row.report_revision,
|
|
)
|
|
)
|
|
if report_record is None:
|
|
continue
|
|
delivery = authorization_cache.get(key)
|
|
if delivery is None:
|
|
delivery = _authorize_execution_delivery(
|
|
session,
|
|
principal,
|
|
registry=registry,
|
|
row=row,
|
|
report_record=report_record,
|
|
)
|
|
authorization_cache[key] = delivery
|
|
payloads.append(
|
|
_execution_payload(
|
|
row,
|
|
report=ReportDefinition.model_validate(report_record.payload),
|
|
registry=registry,
|
|
delivery_authorization=delivery,
|
|
)
|
|
)
|
|
return tuple(payloads)
|
|
|
|
|
|
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,
|
|
run_ref=dataset.source_run_ref,
|
|
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 _enforce_query_access(
|
|
report: ReportDefinition,
|
|
query: ReportQuery,
|
|
) -> ReportQuery:
|
|
policy = report.access_policy
|
|
hidden_dimensions = _policy_strings(policy, "hidden_dimensions")
|
|
hidden_measures = _policy_strings(policy, "hidden_measures")
|
|
requested_dimensions = set(query.dimensions)
|
|
requested_dimensions.update(item.dimension for item in query.filters)
|
|
if query.pivot is not None:
|
|
requested_dimensions.update(query.pivot.rows)
|
|
requested_dimensions.update(query.pivot.columns)
|
|
requested_measures = set(query.measures)
|
|
if query.pivot is not None:
|
|
requested_measures.update(query.pivot.measures)
|
|
blocked = (requested_dimensions & hidden_dimensions) | (
|
|
requested_measures & hidden_measures
|
|
)
|
|
if blocked:
|
|
raise PermissionError(
|
|
"Policy hides requested Reporting fields: "
|
|
+ ", ".join(sorted(blocked))
|
|
)
|
|
if "run" in _policy_strings(policy, "disabled_actions"):
|
|
raise PermissionError(_policy_reason(policy, "run"))
|
|
return query
|
|
|
|
|
|
def _access_explanation(
|
|
report: ReportDefinition,
|
|
query: ReportQuery,
|
|
*,
|
|
source_rows: int,
|
|
authorized_rows: int,
|
|
row_policy: Mapping[str, object],
|
|
) -> dict[str, object]:
|
|
policy = report.access_policy
|
|
hidden_dimensions = sorted(_policy_strings(policy, "hidden_dimensions"))
|
|
hidden_measures = sorted(_policy_strings(policy, "hidden_measures"))
|
|
disabled_actions = sorted(_policy_strings(policy, "disabled_actions"))
|
|
reasons = policy.get("reasons")
|
|
return {
|
|
"hidden_dimensions": hidden_dimensions,
|
|
"hidden_measures": hidden_measures,
|
|
"hidden_rows": max(0, source_rows - authorized_rows),
|
|
"disabled_actions": disabled_actions,
|
|
"reasons": dict(reasons) if isinstance(reasons, Mapping) else {},
|
|
"row_policy": dict(row_policy),
|
|
"effective_query": query.model_dump(mode="json"),
|
|
}
|
|
|
|
|
|
def _authorize_execution_delivery(
|
|
session: Session,
|
|
principal: object,
|
|
*,
|
|
registry: object | None,
|
|
row: ReportingExecution,
|
|
report_record: ReportingDefinitionRecord,
|
|
) -> dict[str, object]:
|
|
report_decision = require_definition_action(
|
|
session,
|
|
principal,
|
|
registry=registry,
|
|
record=report_record,
|
|
action="view",
|
|
)
|
|
semantic_record = get_definition(
|
|
session,
|
|
principal,
|
|
definition_kind="semantic_model",
|
|
definition_id=row.semantic_model_id,
|
|
revision=row.semantic_model_revision,
|
|
)
|
|
dataset_record = get_definition(
|
|
session,
|
|
principal,
|
|
definition_kind="dataset",
|
|
definition_id=row.dataset_id,
|
|
revision=row.dataset_revision,
|
|
)
|
|
if semantic_record is None or dataset_record is None:
|
|
raise PermissionError(
|
|
"The source definitions for this report result are no longer accessible."
|
|
)
|
|
semantic_decision = require_definition_action(
|
|
session,
|
|
principal,
|
|
registry=registry,
|
|
record=semantic_record,
|
|
action="view",
|
|
)
|
|
dataset_decision = require_definition_action(
|
|
session,
|
|
principal,
|
|
registry=registry,
|
|
record=dataset_record,
|
|
action="view",
|
|
)
|
|
dataset = DatasetDefinition.model_validate(dataset_record.payload)
|
|
_empty, row_policy = _apply_row_policy(
|
|
session,
|
|
principal,
|
|
registry=registry,
|
|
dataset_id=dataset_record.definition_id,
|
|
dataset_revision=dataset_record.revision,
|
|
dataset=dataset,
|
|
rows=(),
|
|
)
|
|
return {
|
|
"checked": True,
|
|
"report": report_decision.to_dict(),
|
|
"semantic_model": semantic_decision.to_dict(),
|
|
"dataset": dataset_decision.to_dict(),
|
|
"row_policy": dict(row_policy),
|
|
}
|
|
|
|
|
|
def _policy_strings(policy: Mapping[str, object], key: str) -> set[str]:
|
|
raw = policy.get(key, ())
|
|
if not isinstance(raw, (list, tuple, set, frozenset)):
|
|
return set()
|
|
return {str(item) for item in raw if str(item).strip()}
|
|
|
|
|
|
def _policy_reason(policy: Mapping[str, object], action: str) -> str:
|
|
reasons = policy.get("reasons")
|
|
if isinstance(reasons, Mapping) and str(reasons.get(action) or "").strip():
|
|
return str(reasons[action])
|
|
return f"Policy disables the Reporting {action} action."
|
|
|
|
|
|
def _execution_payload(
|
|
row: ReportingExecution,
|
|
*,
|
|
report: ReportDefinition | None,
|
|
registry: object | None,
|
|
delivery_authorization: Mapping[str, object] | None = 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,
|
|
"delivery_authorization": dict(delivery_authorization or {}),
|
|
}
|
|
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",
|
|
]
|