From 8337aec19ad8430529bc2b2d089286ec38bb6f7a Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Tue, 4 Aug 2026 02:23:24 +0200 Subject: [PATCH] Complete governed reporting execution and publication --- README.md | 9 +- docs/ADMIN_GUIDE.md | 30 ++ docs/REPORTING_BOUNDARY.md | 30 +- docs/USER_GUIDE.md | 22 + src/govoplan_reporting/backend/contracts.py | 4 + src/govoplan_reporting/backend/db/__init__.py | 2 + src/govoplan_reporting/backend/db/models.py | 45 ++ src/govoplan_reporting/backend/definitions.py | 92 +++- src/govoplan_reporting/backend/drilldown.py | 388 +++++++++++++ src/govoplan_reporting/backend/execution.py | 295 +++++++++- src/govoplan_reporting/backend/governance.py | 399 ++++++++++++++ src/govoplan_reporting/backend/manifest.py | 65 ++- .../c8d5e2f6a9b3_reporting_drill_contexts.py | 71 +++ src/govoplan_reporting/backend/operations.py | 50 +- .../backend/postgres_planner.py | 509 ++++++++++++++++++ .../backend/publication_targets.py | 303 +++++++++++ .../backend/query_engine.py | 5 +- src/govoplan_reporting/backend/router.py | 104 +++- src/govoplan_reporting/backend/schemas.py | 51 ++ tests/test_manifest.py | 2 +- tests/test_migrations.py | 1 + tests/test_module_permutations.py | 28 + tests/test_reporting_service.py | 263 +++++++++ webui/scripts/test-interface-pattern.mjs | 3 + webui/src/api/reporting.ts | 146 +++++ .../src/features/reporting/ReportingPage.tsx | 450 +++++++++++++++- .../reporting/ReportingReportsWidget.tsx | 57 ++ webui/src/module.ts | 46 +- webui/src/styles/reporting.css | 222 ++++++++ 29 files changed, 3611 insertions(+), 81 deletions(-) create mode 100644 src/govoplan_reporting/backend/drilldown.py create mode 100644 src/govoplan_reporting/backend/governance.py create mode 100644 src/govoplan_reporting/backend/migrations/versions/c8d5e2f6a9b3_reporting_drill_contexts.py create mode 100644 src/govoplan_reporting/backend/postgres_planner.py create mode 100644 src/govoplan_reporting/backend/publication_targets.py create mode 100644 webui/src/features/reporting/ReportingReportsWidget.tsx diff --git a/README.md b/README.md index 5dbab38..87538c7 100644 --- a/README.md +++ b/README.md @@ -16,15 +16,18 @@ The module now provides an executable governed semantic-reporting vertical: grants, row-policy handoff, freshness checks, and reconstructable run provenance; - safe dimensions, hierarchies, measures, typed calculations, filters, - detail/summary/pivot queries, and accessible chart models without executing - arbitrary report SQL; + detail/summary/pivot queries, parameterized PostgreSQL semantic plans, and + accessible chart models without executing arbitrary report SQL; - quality gates, saved views, interval/scheduled runs, CSV/JSON export, provider-neutral publication targets, and import activation assessments; - a versioned, provider-neutral cross-module report contract with source-owned authorization, declared result schemas, privacy transforms, effective scope, source revisions, purpose, retention, export history, and audit provenance; - a full-height Reporting workspace for running, inspecting, saving, - scheduling, visualizing, and exporting authorized reports. + scheduling, visualizing, drilling into reauthorized contributors, publishing + through Files/Mail, and exporting authorized reports; +- a configurable Dashboard widget and explicit policy explanations for hidden + fields, rows, and actions. Reporting consumes Dataflow outputs or provider-owned read models. It does not read another module's ORM tables or take ownership of ingestion and diff --git a/docs/ADMIN_GUIDE.md b/docs/ADMIN_GUIDE.md index 0a4b9a8..13cc9d4 100644 --- a/docs/ADMIN_GUIDE.md +++ b/docs/ADMIN_GUIDE.md @@ -14,12 +14,28 @@ existing parent revision. Editing creates a new immutable revision and requires the currently observed revision number. Existing runs continue to reference the historical revisions they used. +Each definition also records system, tenant, group, or user governance scope, +whether it is inherited, and whether lower scopes may run, reuse, or automate +it. A child may tighten but never broaden any effective ancestor limit. System +definitions require system governance permission; tenant definitions are bound +to the active tenant; group and user definitions require the matching subject +unless a Reporting administrator performs the operation. Policy is consulted +for view, edit, run, reuse, and automation decisions. + Datasets may bind a static fixture, a pinned Dataflow output, or a capability published by a source-owning module. Do not expose another module's ORM or an unbounded SQL connection as a report source. Configure an explicit schema, freshness policy, source fingerprint expectations, purpose, privacy, retention, and a row-policy provider where source access alone is not enough. +On PostgreSQL installations, Reporting compiles bounded semantic filters, +grouping, measures, calculated measures, ordering, offsets, and limits into a +parameterized PostgreSQL plan over the already authorized provider rows. Field +paths and values are bound parameters and result limits remain mandatory. Pivot +plans retain the safe provider-neutral engine fallback. SQLite development and +other database engines use the same typed semantics through the bounded runtime +engine. + ## Access and publication Tenant-visible definitions are readable by principals with Reporting read @@ -32,6 +48,20 @@ the Reporting publication-target contract. The target receives one immutable execution payload and an idempotency key. It must return bounded evidence and must not expose credentials in that evidence. +Reporting ships two optional adapters. `reporting.publication.files` calls +`files.artifact_store` and stores an idempotent managed artifact with execution, +revision, output-hash, and file-version evidence. `reporting.publication.mail` +calls `mail.notificationDelivery` and submits an idempotent report notice to the +Mail outbox. The latter does not bypass Mail profile, credential, or transport +policy. Adapter availability is evaluated at runtime, so Reporting remains +usable with neither Files nor Mail installed. + +Drill contexts expire after 20 minutes, are bound to the creating account, store +only token and context hashes, and must match the original execution output and +source fingerprints. Resolution re-runs definition and row-level authorization. +Treat a fingerprint mismatch as a required report rerun, not as a recoverable +client warning. + ## Cross-module provider governance Source modules register `reporting.report_provider.` capabilities; diff --git a/docs/REPORTING_BOUNDARY.md b/docs/REPORTING_BOUNDARY.md index 6983fe2..876a08f 100644 --- a/docs/REPORTING_BOUNDARY.md +++ b/docs/REPORTING_BOUNDARY.md @@ -83,6 +83,10 @@ Reporting does not own: outcomes. - `reporting.chart_renderer` renders provider-neutral visual models with an accessible table fallback. +- `reporting.publication.files` adapts immutable results to Core's + `files.artifact_store` boundary without importing Files internals. +- `reporting.publication.mail` adapts report notices to Core's + `mail.notificationDelivery` boundary without importing Mail internals. - `reporting.read_model:*` capabilities can expose bounded source-owned rows. - `reporting.publication_target:*` capabilities can accept immutable result payloads without Reporting importing the target module. @@ -125,14 +129,22 @@ row-policy provenance, blocking quality plans, definition hashes, executor version, output hash, diagnostics, and authorized rows are retained with the execution. Failed runs also retain evidence. -The query engine deliberately implements a typed expression and semantic +The query layer deliberately implements a typed expression and semantic query language rather than `eval`, arbitrary SQL, stored procedures, or -runtime scripts. It supports detail, grouped summary, pivot, dimensions, +runtime scripts. PostgreSQL installations receive parameterized semantic plans +for filters, grouping, measures, calculated aggregates, sorting, and bounds; +other engines and pivots use the equivalent bounded runtime evaluator. It +supports detail, grouped summary, pivot, dimensions, hierarchies, common aggregates, calculated measures, filters, sorting, pagination, totals, and a provider-neutral visualization model. A saved chart that is incompatible with an ad-hoc query degrades to its mandatory table fallback instead of failing a valid report run. +Aggregate drill-through uses an expiring actor-bound context hash. Resolution +rechecks all definition and row-policy decisions, verifies the source +fingerprints against the original execution, preserves the complete dimension +path, and returns only authorized contributors. + Direct export supports UTF-8 CSV and JSON. CSV cells that spreadsheet software could interpret as formulas are escaped. Additional formats and delivery destinations use an optional publication capability and preserve idempotent @@ -142,7 +154,10 @@ unsupported executable behavior remains. The WebUI uses the platform module loader and common controls. It exposes a report catalogue, parameter and semantic-query controls, result visualization -and table views, history/provenance, saved views, schedules, and downloads. +and table views, accessible bar/column/line/area/pie/donut/metric charts, +drill-through, access explanations, history/provenance, saved views, schedule +management, Files/Mail publication management, downloads, and a Dashboard +widget contribution. The global `/reports` route is owned only by Reporting. `/reporting` is a documented compatibility path. Campaign's module-local aggregate view remains at `/campaigns/reports`; when both modules are enabled, the same safe aggregate @@ -151,7 +166,8 @@ contract. ## Remaining Product Depth -The architecture boundary is implemented. Further work is additive product -depth: richer visualization providers, drill-through navigation, packaged -domain report catalogues, XLSX/PDF formatting through optional providers, and -target-environment evidence for a maturity claim above `vertical_slice`. +The architecture boundary and first operational vertical are implemented. +Further work is additive product depth: packaged domain report catalogues, +XLSX/PDF formatting through optional renderer providers, selector-backed Mail +profile configuration, external publication connectors, and target-environment +evidence for a maturity claim above `vertical_slice`. diff --git a/docs/USER_GUIDE.md b/docs/USER_GUIDE.md index e09cf41..0951e74 100644 --- a/docs/USER_GUIDE.md +++ b/docs/USER_GUIDE.md @@ -13,6 +13,13 @@ engine. A successful result shows its authorized row count, visualization, and table. When a saved chart does not match an ad-hoc query, Reporting shows the accessible table fallback instead of changing or rejecting the query. +Summary and pivot rows expose a detail action. Selecting it creates a short-lived, +account-bound drill context, rechecks the report, semantic model, dataset, row +policy, and source fingerprint, and then displays only the authorized contributing +rows. The path above the table records every aggregate dimension used for the +drill. If the source changed, run the report again rather than treating stale +aggregate and detail states as equivalent. + ## Inspect evidence The right panel lists previous runs and definition/source pins. Select an @@ -21,6 +28,10 @@ the exact definition and output. Warnings explain freshness, inferred schema, or provider diagnostics. A failed quality gate records a failed execution and does not publish a result. +The **Effective access** explanation states when dimensions, measures, source +rows, or actions were removed by Policy. A result with no hidden elements says +so explicitly; catalogue visibility never grants access to protected detail. + ## Save and export Use **Save current view** to keep the current query under your account. Saved @@ -32,6 +43,17 @@ Users with scheduling permission can create an hourly, daily, weekly, or 30-day interval from the current revision, parameters, and query. Scheduled runs continue to use those exact pins until the schedule is edited. +The Schedules panel can pause or resume each schedule with optimistic revision +checking. Users with publication permission can publish a successful execution +to Files or Mail. Files stores CSV, JSON, or accessible HTML through managed +artifact storage. Mail submits a bounded report notice through its durable +outbox and requires a usable profile, sender, and recipient. Unavailable targets +remain explained but cannot be selected as a valid destination. Publication +history records the target, result, time, output hash, and provider evidence. + +When Dashboard is enabled, the **Reports** widget lists active reports without +copying result data into Dashboard. Its item limit is configurable per widget. + ## Run a module report Module reports retain their source module's access rules. Select the source diff --git a/src/govoplan_reporting/backend/contracts.py b/src/govoplan_reporting/backend/contracts.py index c32517e..820fc33 100644 --- a/src/govoplan_reporting/backend/contracts.py +++ b/src/govoplan_reporting/backend/contracts.py @@ -10,6 +10,8 @@ CAPABILITY_REPORTING_REGISTRY = "reporting.registry" CAPABILITY_REPORTING_RUNNER = "reporting.runner" CAPABILITY_REPORTING_SCHEDULER = "reporting.scheduler" CAPABILITY_REPORTING_CHART_RENDERER = "reporting.chart_renderer" +CAPABILITY_REPORTING_PUBLICATION_FILES = "reporting.publication.files" +CAPABILITY_REPORTING_PUBLICATION_MAIL = "reporting.publication.mail" @dataclass(frozen=True, slots=True) @@ -119,6 +121,8 @@ def capability(registry: object | None, name: str) -> object | None: __all__ = [ "CAPABILITY_REPORTING_CHART_RENDERER", + "CAPABILITY_REPORTING_PUBLICATION_FILES", + "CAPABILITY_REPORTING_PUBLICATION_MAIL", "CAPABILITY_REPORTING_REGISTRY", "CAPABILITY_REPORTING_RUNNER", "CAPABILITY_REPORTING_SCHEDULER", diff --git a/src/govoplan_reporting/backend/db/__init__.py b/src/govoplan_reporting/backend/db/__init__.py index ed66639..1bfe9fe 100644 --- a/src/govoplan_reporting/backend/db/__init__.py +++ b/src/govoplan_reporting/backend/db/__init__.py @@ -4,6 +4,7 @@ from govoplan_reporting.backend.db.models import ( ReportingDefinitionGrant, ReportingDefinitionIdentity, ReportingDefinitionRevision, + ReportingDrillContext, ReportingExecution, ReportingImportAssessment, ReportingPublication, @@ -16,6 +17,7 @@ __all__ = [ "ReportingDefinitionGrant", "ReportingDefinitionIdentity", "ReportingDefinitionRevision", + "ReportingDrillContext", "ReportingExecution", "ReportingImportAssessment", "ReportingPublication", diff --git a/src/govoplan_reporting/backend/db/models.py b/src/govoplan_reporting/backend/db/models.py index 140dfe5..b805552 100644 --- a/src/govoplan_reporting/backend/db/models.py +++ b/src/govoplan_reporting/backend/db/models.py @@ -484,6 +484,50 @@ class ReportingPublication(Base, TimestampMixin): ) +class ReportingDrillContext(Base, TimestampMixin): + __tablename__ = "reporting_drill_contexts" + __table_args__ = ( + UniqueConstraint( + "tenant_id", "drill_context_id", name="uq_reporting_drill_context" + ), + Index( + "ix_reporting_drill_context_expiry", + "tenant_id", + "expires_at", + ), + Index( + "ix_reporting_drill_context_execution", + "tenant_id", + "execution_id", + ), + ) + + id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid) + tenant_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True) + drill_context_id: Mapped[str] = mapped_column( + String(36), nullable=False, index=True + ) + execution_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True) + token_sha256: Mapped[str] = mapped_column(String(64), nullable=False) + context_sha256: Mapped[str] = mapped_column(String(64), nullable=False) + actor_id: Mapped[str] = mapped_column(String(255), nullable=False, index=True) + dimension_path: Mapped[list[dict[str, Any]]] = mapped_column( + JSON, default=list, nullable=False + ) + source_fingerprints: Mapped[list[dict[str, Any]]] = mapped_column( + JSON, default=list, nullable=False + ) + policy_provenance: Mapped[dict[str, Any]] = mapped_column( + JSON, default=dict, nullable=False + ) + expires_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, index=True + ) + last_accessed_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), nullable=True + ) + + class ReportingQualityResult(Base, TimestampMixin): __tablename__ = "reporting_quality_results" __table_args__ = ( @@ -551,6 +595,7 @@ __all__ = [ "ReportingDefinitionGrant", "ReportingDefinitionIdentity", "ReportingDefinitionRevision", + "ReportingDrillContext", "ReportingExecution", "ReportingImportAssessment", "ReportingPublication", diff --git a/src/govoplan_reporting/backend/definitions.py b/src/govoplan_reporting/backend/definitions.py index a49f0f3..41589af 100644 --- a/src/govoplan_reporting/backend/definitions.py +++ b/src/govoplan_reporting/backend/definitions.py @@ -27,6 +27,11 @@ from govoplan_reporting.backend.domain import ( ReportingDefinitionRecord, definition_from_row, ) +from govoplan_reporting.backend.governance import ( + apply_parent_governance, + normalize_definition_governance, + scope_visible, +) from govoplan_reporting.backend.schemas import validate_definition_payload @@ -86,17 +91,29 @@ def create_definition( clean_reason = _required(change_reason, "Reporting change reason", 1_000) _aware(recorded_at, "Reporting recorded_at") validated_payload = validate_definition_payload(kind, dict(payload)) + validated_payload = validate_definition_payload( + kind, + normalize_definition_governance( + validated_payload, + principal, + administrative=_has_scope(principal, ADMIN_SCOPE), + ), + ) parent_kind, parent_id, parent_revision = _parent_reference( kind, validated_payload, ) - _validate_parent( - session, - tenant_id=tenant_id, - child_status=clean_status, - parent_kind=parent_kind, - parent_id=parent_id, - parent_revision=parent_revision, + validated_payload = validate_definition_payload( + kind, + _validate_parent( + session, + tenant_id=tenant_id, + child_status=clean_status, + parent_kind=parent_kind, + parent_id=parent_id, + parent_revision=parent_revision, + child_payload=validated_payload, + ), ) request = { "definition_kind": kind, @@ -241,14 +258,26 @@ def update_definition( kind, dict(changes.get("payload", current.payload)), ) + next_payload = validate_definition_payload( + kind, + normalize_definition_governance( + next_payload, + principal, + administrative=_has_scope(principal, ADMIN_SCOPE), + ), + ) parent_kind, parent_id, parent_revision = _parent_reference(kind, next_payload) - _validate_parent( - session, - tenant_id=tenant_id, - child_status=next_status, - parent_kind=parent_kind, - parent_id=parent_id, - parent_revision=parent_revision, + next_payload = validate_definition_payload( + kind, + _validate_parent( + session, + tenant_id=tenant_id, + child_status=next_status, + parent_kind=parent_kind, + parent_id=parent_id, + parent_revision=parent_revision, + child_payload=next_payload, + ), ) identity = _identity(session, tenant_id, kind, definition_id) if identity is None: @@ -584,9 +613,10 @@ def _validate_parent( parent_kind: str | None, parent_id: str | None, parent_revision: int | None, -) -> None: + child_payload: Mapping[str, object], +) -> dict[str, object]: if parent_kind is None: - return + return dict(child_payload) row = ( session.query(ReportingDefinitionRevision) .filter( @@ -605,6 +635,7 @@ def _validate_parent( raise ReportingDefinitionError( f"An active Reporting definition requires an active {parent_kind} revision." ) + return apply_parent_governance(child_payload, row.payload) def _parent_reference( @@ -667,6 +698,8 @@ def _can_access( ) if row is None: return False + if not scope_visible(row.payload, principal): + return False if _has_scope(principal, ADMIN_SCOPE): return True identity = _identity(session, tenant_id, definition_kind, definition_id) @@ -712,6 +745,24 @@ def _require_scope(principal: object, scope: str) -> None: def _filter_accessible(query: Query, principal: object) -> Query: if _has_scope(principal, ADMIN_SCOPE): return query + governance = ReportingDefinitionRevision.payload["governance"] + scope_type = governance["scope_type"].as_string() + scope_id = governance["scope_id"].as_string() + inherited = governance["inherit_to_lower_scopes"].as_boolean() + scope_conditions = [ + scope_type.is_(None), + and_( + scope_type == "tenant", + or_(scope_id.is_(None), scope_id == _principal_tenant(principal)), + ), + and_(scope_type == "system", inherited.is_(True)), + ] + group_ids = tuple(_string_subject_ids(principal, "group_ids")) + if group_ids: + scope_conditions.append(and_(scope_type == "group", scope_id.in_(group_ids))) + user_ids = _actor_ids(principal) + if user_ids: + scope_conditions.append(and_(scope_type == "user", scope_id.in_(user_ids))) conditions = [ReportingDefinitionRevision.visibility == "tenant"] actor_ids = _actor_ids(principal) if actor_ids: @@ -752,7 +803,14 @@ def _filter_accessible(query: Query, principal: object) -> Query: ) ) ) - return query.filter(or_(*conditions)) + return query.filter(and_(or_(*scope_conditions), or_(*conditions))) + + +def _string_subject_ids(principal: object, attribute: str) -> tuple[str, ...]: + raw = getattr(principal, attribute, ()) or () + if isinstance(raw, (str, bytes)): + return (str(raw),) if raw else () + return tuple(dict.fromkeys(str(value) for value in raw if str(value or "").strip())) def _current_row( diff --git a/src/govoplan_reporting/backend/drilldown.py b/src/govoplan_reporting/backend/drilldown.py new file mode 100644 index 0000000..52deab0 --- /dev/null +++ b/src/govoplan_reporting/backend/drilldown.py @@ -0,0 +1,388 @@ +from __future__ import annotations + +from collections.abc import Mapping, Sequence +from datetime import UTC, datetime, timedelta +import hashlib +import hmac +import json +import secrets +from typing import Any +import uuid + +from sqlalchemy.orm import Session + +from govoplan_core.security.time import utc_now +from govoplan_reporting.backend.db.models import ( + ReportingDrillContext, + ReportingExecution, +) +from govoplan_reporting.backend.definitions import get_definition +from govoplan_reporting.backend.execution import ( + ReportingExecutionError, + _apply_row_policy, + _read_dataset, + _validate_schema, + get_execution, +) +from govoplan_reporting.backend.postgres_planner import execute_postgres_query +from govoplan_reporting.backend.query_engine import execute_semantic_query +from govoplan_reporting.backend.schemas import ( + DatasetDefinition, + FilterClause, + ReportDefinition, + ReportQuery, + SemanticModelDefinition, +) + + +DRILL_CONTEXT_TTL = timedelta(minutes=20) + + +class ReportingDrillError(ValueError): + pass + + +def create_drill_context( + session: Session, + principal: object, + *, + registry: object | None, + execution_id: str, + aggregate_row: Mapping[str, object], + limit: int, +) -> dict[str, object]: + execution_payload = get_execution( + session, + principal, + execution_id=execution_id, + registry=registry, + ) + if execution_payload is None or execution_payload.get("status") != "succeeded": + raise LookupError("Successful Reporting execution not found.") + row = _execution(session, _tenant(principal), execution_id) + normalized_aggregate = _json_value(dict(aggregate_row)) + if normalized_aggregate not in [ + _json_value(dict(item)) for item in row.result_rows or [] + ]: + raise ReportingDrillError( + "The selected aggregate row does not belong to this execution." + ) + semantic_record = get_definition( + session, + principal, + definition_kind="semantic_model", + definition_id=row.semantic_model_id, + revision=row.semantic_model_revision, + ) + if semantic_record is None: + raise PermissionError("The report semantic model is no longer accessible.") + semantic = SemanticModelDefinition.model_validate(semantic_record.payload) + query = ReportQuery.model_validate(row.query or {}) + dimension_keys = _drill_dimensions(query, semantic) + dimension_map = {item.key: item for item in semantic.dimensions} + path = [ + { + "dimension": key, + "label": dimension_map[key].label, + "value": normalized_aggregate.get(key), + } + for key in dimension_keys + if key in normalized_aggregate + ] + if not path: + raise ReportingDrillError( + "This aggregate has no dimension path to drill through." + ) + bounded_limit = max(1, min(int(limit), 500)) + actor_id = _actor(principal) + if not actor_id: + raise ReportingDrillError("Drill-through requires an accountable actor.") + drill_context_id = str(uuid.uuid4()) + secret = secrets.token_urlsafe(32) + token = f"{drill_context_id}.{secret}" + context = { + "execution_id": execution_id, + "output_hash": row.output_hash, + "actor_id": actor_id, + "dimension_path": path, + "source_fingerprints": list(row.source_fingerprints or []), + "limit": bounded_limit, + } + item = ReportingDrillContext( + tenant_id=row.tenant_id, + drill_context_id=drill_context_id, + execution_id=execution_id, + token_sha256=_sha256(token), + context_sha256=_sha256(context), + actor_id=actor_id, + dimension_path=path, + source_fingerprints=list(row.source_fingerprints or []), + policy_provenance=dict( + execution_payload.get("delivery_authorization") or {} + ), + expires_at=utc_now() + DRILL_CONTEXT_TTL, + ) + item.policy_provenance["limit"] = bounded_limit + session.add(item) + session.flush() + return { + "token": token, + "drill_context_id": drill_context_id, + "execution_id": execution_id, + "dimension_path": path, + "expires_at": _datetime_text(item.expires_at), + } + + +def resolve_drill_context( + session: Session, + principal: object, + *, + registry: object | None, + token: str, +) -> dict[str, object]: + context_id, separator, _secret = token.partition(".") + if not separator or not context_id: + raise ReportingDrillError("The drill-through context token is invalid.") + item = ( + session.query(ReportingDrillContext) + .filter( + ReportingDrillContext.tenant_id == _tenant(principal), + ReportingDrillContext.drill_context_id == context_id, + ) + .one_or_none() + ) + if item is None or not hmac.compare_digest(item.token_sha256, _sha256(token)): + raise LookupError("Reporting drill-through context not found.") + if item.actor_id != _actor(principal): + raise PermissionError( + "This drill-through context belongs to another account." + ) + if _aware(item.expires_at) <= utc_now(): + raise ReportingDrillError("The drill-through context has expired.") + row = _execution(session, item.tenant_id, item.execution_id) + expected_context = { + "execution_id": row.execution_id, + "output_hash": row.output_hash, + "actor_id": item.actor_id, + "dimension_path": list(item.dimension_path or []), + "source_fingerprints": list(item.source_fingerprints or []), + "limit": int((item.policy_provenance or {}).get("limit", 200)), + } + if not hmac.compare_digest(item.context_sha256, _sha256(expected_context)): + raise ReportingDrillError( + "The persisted drill-through context failed its integrity check." + ) + execution_payload = get_execution( + session, + principal, + execution_id=row.execution_id, + registry=registry, + ) + if execution_payload is None: + raise LookupError("Reporting execution not found.") + report_record = get_definition( + session, + principal, + definition_kind="report", + definition_id=row.report_id, + revision=row.report_revision, + ) + 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 report_record is None or semantic_record is None or dataset_record is None: + raise PermissionError( + "The report source graph is no longer accessible for drill-through." + ) + report = ReportDefinition.model_validate(report_record.payload) + semantic = SemanticModelDefinition.model_validate(semantic_record.payload) + dataset = DatasetDefinition.model_validate(dataset_record.payload) + source = _read_dataset( + session, + principal, + registry=registry, + dataset=dataset, + parameters=dict(row.parameters or {}), + ) + if not _fingerprints_equal( + item.source_fingerprints or [], source.source_fingerprints + ): + raise ReportingExecutionError( + "The source fingerprint changed after the aggregate execution; run the report again before drilling through." + ) + normalized_rows = tuple(_json_value(dict(source_row)) for source_row in source.rows) + _validate_schema(dataset, normalized_rows) + authorized_rows, row_policy = _apply_row_policy( + session, + principal, + registry=registry, + dataset_id=dataset_record.definition_id, + dataset_revision=dataset_record.revision, + dataset=dataset, + rows=normalized_rows, + ) + original = ReportQuery.model_validate(row.query or {}) + hidden_dimensions = _strings(report.access_policy.get("hidden_dimensions")) + visible_dimensions = [ + dimension.key + for dimension in semantic.dimensions + if dimension.key not in hidden_dimensions + ] + filters = list(original.filters) + filters.extend( + FilterClause( + dimension=str(path_item["dimension"]), + operator="eq", + value=path_item.get("value"), + ) + for path_item in item.dimension_path or [] + ) + detail_query = ReportQuery( + mode="detail", + dimensions=visible_dimensions, + filters=filters, + limit=int((item.policy_provenance or {}).get("limit", 200)), + ) + result = execute_postgres_query( + session, + rows=authorized_rows, + dataset=dataset, + semantic_model=semantic, + query=detail_query, + ) or execute_semantic_query(authorized_rows, semantic, detail_query) + item.last_accessed_at = utc_now() + item.policy_provenance = { + **dict(item.policy_provenance or {}), + "resolved_row_policy": dict(row_policy), + "delivery_authorization": dict( + execution_payload.get("delivery_authorization") or {} + ), + } + session.flush() + return { + "drill_context_id": item.drill_context_id, + "execution_id": item.execution_id, + "dimension_path": list(item.dimension_path or []), + "rows": list(result.rows), + "schema": list(result.schema), + "total_rows": result.total_rows, + "truncated": result.truncated or source.truncated, + "source_fingerprints": list(source.source_fingerprints), + "policy_provenance": dict(item.policy_provenance or {}), + "expires_at": _datetime_text(item.expires_at), + } + + +def _drill_dimensions( + query: ReportQuery, + semantic: SemanticModelDefinition, +) -> tuple[str, ...]: + if query.mode == "pivot" and query.pivot is not None: + return tuple(dict.fromkeys((*query.pivot.rows, *query.pivot.columns))) + return tuple(query.dimensions or semantic.default_dimensions) + + +def _execution( + session: Session, + tenant_id: str, + execution_id: str, +) -> ReportingExecution: + row = ( + session.query(ReportingExecution) + .filter( + ReportingExecution.tenant_id == tenant_id, + ReportingExecution.execution_id == execution_id, + ) + .one_or_none() + ) + if row is None: + raise LookupError("Reporting execution not found.") + return row + + +def _fingerprints_equal( + expected: Sequence[Mapping[str, object]], + actual: Sequence[Mapping[str, object]], +) -> bool: + normalize = lambda values: sorted( # noqa: E731 - compact canonicalizer + json.dumps( + _json_value(dict(item)), + sort_keys=True, + separators=(",", ":"), + ensure_ascii=True, + ) + for item in values + ) + return normalize(expected) == normalize(actual) + + +def _tenant(principal: object) -> str: + tenant_id = str(getattr(principal, "tenant_id", "") or "").strip() + if not tenant_id: + raise ReportingDrillError("Drill-through 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 _strings(value: object) -> set[str]: + if not isinstance(value, (list, tuple, set, frozenset)): + return set() + return {str(item) for item in value if str(item).strip()} + + +def _sha256(value: object) -> str: + payload = value if isinstance(value, str) else json.dumps( + _json_value(value), + sort_keys=True, + separators=(",", ":"), + ensure_ascii=True, + ) + return hashlib.sha256(payload.encode("utf-8")).hexdigest() + + +def _json_value(value: object) -> Any: + if isinstance(value, datetime): + return _aware(value).isoformat() + 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 _aware(value: datetime) -> datetime: + return value if value.tzinfo is not None else value.replace(tzinfo=UTC) + + +def _datetime_text(value: datetime) -> str: + return _aware(value).isoformat() + + +__all__ = [ + "DRILL_CONTEXT_TTL", + "ReportingDrillError", + "create_drill_context", + "resolve_drill_context", +] diff --git a/src/govoplan_reporting/backend/execution.py b/src/govoplan_reporting/backend/execution.py index 0a0b3a6..8d8544a 100644 --- a/src/govoplan_reporting/backend/execution.py +++ b/src/govoplan_reporting/backend/execution.py @@ -37,6 +37,12 @@ from govoplan_reporting.backend.db.models import ( 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, @@ -99,7 +105,12 @@ class SqlReportingRunner: *, execution_id: str, ) -> Mapping[str, object] | None: - return get_execution(_session(session), principal, execution_id=execution_id) + return get_execution( + _session(session), + principal, + execution_id=execution_id, + registry=self.registry, + ) def execute_report( @@ -126,6 +137,13 @@ def execute_report( 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, @@ -138,6 +156,13 @@ def execute_report( "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, @@ -150,8 +175,15 @@ def execute_report( "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 = query or report.default_query + effective_query = _enforce_query_access(report, query or report.default_query) clean_idempotency_key = _required( idempotency_key, "Reporting execution idempotency key", @@ -176,7 +208,19 @@ def execute_report( request_sha256=request_sha256, ) if replay is not None: - return _execution_payload(replay, report=report, registry=registry) + 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), @@ -232,7 +276,14 @@ def execute_report( output_hash=source.output_hash, source_fingerprints=source.source_fingerprints, ) - result = execute_semantic_query(authorized_rows, semantic, effective_query) + 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, @@ -244,7 +295,12 @@ def execute_report( 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}" + 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 @@ -259,11 +315,35 @@ def execute_report( "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) - return _execution_payload(execution, report=report, registry=registry) + 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() @@ -284,6 +364,7 @@ def get_execution( principal: object, *, execution_id: str, + registry: object | None = None, ) -> dict[str, object] | None: row = ( session.query(ReportingExecution) @@ -304,10 +385,18 @@ def get_execution( ) 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=None, + registry=registry, + delivery_authorization=delivery, ) @@ -317,16 +406,15 @@ def list_executions( *, report_id: str, limit: int = 100, + registry: object | None = None, ) -> tuple[dict[str, object], ...]: - if ( - get_definition( - session, - principal, - definition_kind="report", - definition_id=report_id, - ) - is None - ): + current_report = get_definition( + session, + principal, + definition_kind="report", + definition_id=report_id, + ) + if current_report is None: return () rows = ( session.query(ReportingExecution) @@ -338,7 +426,46 @@ def list_executions( .limit(max(1, min(limit, 200))) .all() ) - return tuple(_execution_payload(row, report=None, registry=None) for row in rows) + 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( @@ -717,11 +844,144 @@ def _evaluate_assertion( } +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, @@ -747,6 +1007,7 @@ def _execution_payload( "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( diff --git a/src/govoplan_reporting/backend/governance.py b/src/govoplan_reporting/backend/governance.py new file mode 100644 index 0000000..8f01a3f --- /dev/null +++ b/src/govoplan_reporting/backend/governance.py @@ -0,0 +1,399 @@ +from __future__ import annotations + +from collections.abc import Mapping +from typing import Literal, cast + +from govoplan_core.core.access import PrincipalRef +from govoplan_core.core.policy import ( + DefinitionGovernanceAction, + DefinitionGovernanceRequest, + DefinitionScopeRef, + PolicyDecision, + PolicySourceStep, + definition_governance_policy, +) +from govoplan_reporting.backend.domain import ReportingDefinitionRecord +from govoplan_reporting.backend.schemas import DefinitionGovernance + + +_LIMITS = ( + "inherit_to_lower_scopes", + "allow_run", + "allow_reuse", + "allow_automation", +) +_SCOPE_RANK = {"system": 0, "tenant": 1, "group": 2, "user": 3} + + +class ReportingGovernanceError(ValueError): + pass + + +def normalize_definition_governance( + payload: Mapping[str, object], + principal: object, + *, + administrative: bool, +) -> dict[str, object]: + result = dict(payload) + raw = result.get("governance") + governance = DefinitionGovernance.model_validate( + raw if isinstance(raw, Mapping) else {} + ) + scope_type = governance.scope_type + scope_id = str(governance.scope_id or "").strip() or None + tenant_id = _tenant(principal) + if scope_type == "system": + if not _has_scope(principal, "system:governance:write"): + raise PermissionError( + "System Reporting definitions require system governance permission." + ) + elif scope_type == "tenant": + if scope_id not in {None, tenant_id}: + raise PermissionError( + "Reporting definitions can only target the active tenant." + ) + scope_id = tenant_id + elif scope_type == "group": + if scope_id not in _string_set(getattr(principal, "group_ids", ())): + if not administrative: + raise PermissionError( + "Group Reporting definitions require membership in that group." + ) + elif scope_type == "user": + own_ids = { + str(getattr(principal, "account_id", "") or ""), + str(getattr(principal, "membership_id", "") or ""), + } + if scope_id not in own_ids and not administrative: + raise PermissionError( + "User Reporting definitions can only target the current account." + ) + if scope_id == str(getattr(principal, "membership_id", "") or ""): + scope_id = str(getattr(principal, "account_id", "") or "") + effective = _effective_limits(governance) + result["governance"] = governance.model_copy( + update={ + "scope_id": scope_id, + "inherit_to_lower_scopes": effective["inherit_to_lower_scopes"], + "allow_run": effective["allow_run"], + "allow_reuse": effective["allow_reuse"], + "allow_automation": effective["allow_automation"], + "source_effective_limits": dict(effective), + } + ).model_dump(mode="json") + return result + + +def validate_parent_governance( + child_payload: Mapping[str, object], + parent_payload: Mapping[str, object], +) -> None: + child = _governance(child_payload) + parent = _governance(parent_payload) + child_scope = _scope(child) + parent_scope = _scope(parent) + if _SCOPE_RANK[child_scope.scope_type] < _SCOPE_RANK[parent_scope.scope_type]: + raise ReportingGovernanceError( + "A Reporting definition cannot broaden the scope of its parent." + ) + if child_scope != parent_scope and not parent.inherit_to_lower_scopes: + raise ReportingGovernanceError( + "The parent Reporting definition is not inherited by lower scopes." + ) + parent_limits = _effective_limits(parent) + child_limits = _effective_limits(child) + broadened = [key for key in _LIMITS if child_limits[key] and not parent_limits[key]] + if broadened: + raise ReportingGovernanceError( + "A child Reporting definition cannot broaden inherited limits: " + + ", ".join(sorted(broadened)) + ) + + +def apply_parent_governance( + child_payload: Mapping[str, object], + parent_payload: Mapping[str, object], +) -> dict[str, object]: + """Persist the effective parent restriction and its immediate provenance.""" + + validate_parent_governance(child_payload, parent_payload) + child = _governance(child_payload) + parent = _governance(parent_payload) + parent_limits = _effective_limits(parent) + effective = { + key: bool(getattr(child, key)) and parent_limits[key] for key in _LIMITS + } + parent_scope = { + "scope_type": parent.scope_type, + "scope_id": parent.scope_id, + } + if parent.source_scope: + parent_scope["inherited_from"] = dict(parent.source_scope) + result = dict(child_payload) + result["governance"] = child.model_copy( + update={ + "inherit_to_lower_scopes": effective["inherit_to_lower_scopes"], + "allow_run": effective["allow_run"], + "allow_reuse": effective["allow_reuse"], + "allow_automation": effective["allow_automation"], + "source_scope": parent_scope, + "source_effective_limits": effective, + "derivation_provenance": { + **dict(child.derivation_provenance), + "parent_scope": parent_scope, + "restriction_mode": "intersection", + }, + } + ).model_dump(mode="json") + return result + + +def definition_decision( + session: object, + principal: object, + *, + registry: object | None, + record: ReportingDefinitionRecord, + action: DefinitionGovernanceAction, +) -> PolicyDecision: + governance = _governance(record.payload) + source = _scope(governance) + target = _target_scope(source, principal) + request = DefinitionGovernanceRequest( + module_id="reporting", + definition_ref=f"{record.definition_kind}:{record.definition_id}:{record.revision}", + tenant_id=_tenant(principal), + definition_scope=source, + target_scope=target, + definition_kind=cast(Literal["flow", "template"], "flow"), + action=action, + actor=_principal_ref(principal), + status=record.status, + inherit_to_lower_scopes=governance.inherit_to_lower_scopes, + allow_run=governance.allow_run, + allow_reuse=governance.allow_reuse, + allow_automation=governance.allow_automation, + context={ + "ancestor_limits": dict(governance.source_effective_limits), + "ancestor_source": dict(governance.source_scope or {}), + "reporting_definition_kind": record.definition_kind, + }, + ) + provider = definition_governance_policy(registry) + if provider is not None: + return provider.resolve_definition_action(session, request=request) + return _fallback_decision(request) + + +def require_definition_action( + session: object, + principal: object, + *, + registry: object | None, + record: ReportingDefinitionRecord, + action: DefinitionGovernanceAction, +) -> PolicyDecision: + decision = definition_decision( + session, + principal, + registry=registry, + record=record, + action=action, + ) + if not decision.allowed: + raise PermissionError( + decision.reason or f"Reporting definition action is denied: {action}." + ) + return decision + + +def governance_payload(payload: Mapping[str, object]) -> dict[str, object]: + governance = _governance(payload) + return { + **governance.model_dump(mode="json"), + "effective_limits": _effective_limits(governance), + } + + +def scope_visible(payload: Mapping[str, object], principal: object) -> bool: + governance = _governance(payload) + scope = _scope(governance) + if scope.scope_type == "system": + return governance.inherit_to_lower_scopes or _has_scope( + principal, "reporting:definition:admin" + ) + if scope.scope_type == "tenant": + return scope.scope_id in {None, _tenant(principal)} + if scope.scope_type == "group": + return scope.scope_id in _string_set(getattr(principal, "group_ids", ())) + return scope.scope_id in { + str(getattr(principal, "account_id", "") or ""), + str(getattr(principal, "membership_id", "") or ""), + } + + +def _fallback_decision(request: DefinitionGovernanceRequest) -> PolicyDecision: + source = request.definition_scope + target = request.target_scope + same_scope = source == target + inherited = ( + _SCOPE_RANK[target.scope_type] >= _SCOPE_RANK[source.scope_type] + and request.inherit_to_lower_scopes + ) + visible = same_scope or inherited + if request.action == "view": + allowed = visible + elif request.action == "edit": + allowed = same_scope + elif request.action == "run": + allowed = visible and request.status == "active" and request.allow_run + elif request.action == "reuse": + allowed = visible and request.allow_reuse + elif request.action == "automate": + allowed = visible and request.allow_automation + else: + allowed = visible and request.allow_reuse + reason = ( + None + if allowed + else ( + "The Reporting definition's scope or inherited limits do not allow this action." + ) + ) + return PolicyDecision( + allowed=allowed, + reason=reason, + source_path=( + PolicySourceStep( + scope_type=source.scope_type, + scope_id=source.scope_id, + label="Reporting definition governance", + applied_fields=_LIMITS, + policy={ + "inherit_to_lower_scopes": request.inherit_to_lower_scopes, + "allow_run": request.allow_run, + "allow_reuse": request.allow_reuse, + "allow_automation": request.allow_automation, + }, + ), + ), + requirements=() if allowed else (f"reporting.definition.{request.action}",), + details={ + "provider": "reporting.conservative_fallback", + "definition_scope": source.path, + "target_scope": target.path, + "action": request.action, + }, + ) + + +def _governance(payload: Mapping[str, object]) -> DefinitionGovernance: + raw = payload.get("governance") + return DefinitionGovernance.model_validate(raw if isinstance(raw, Mapping) else {}) + + +def _scope(governance: DefinitionGovernance) -> DefinitionScopeRef: + return DefinitionScopeRef( + scope_type=governance.scope_type, + scope_id=governance.scope_id, + ) + + +def _target_scope(source: DefinitionScopeRef, principal: object) -> DefinitionScopeRef: + if source.scope_type == "group" and source.scope_id in _string_set( + getattr(principal, "group_ids", ()) + ): + return source + own_ids = { + str(getattr(principal, "account_id", "") or ""), + str(getattr(principal, "membership_id", "") or ""), + } + if source.scope_type == "user" and source.scope_id in own_ids: + return source + return DefinitionScopeRef("tenant", _tenant(principal)) + + +def _effective_limits(governance: DefinitionGovernance) -> dict[str, bool]: + source = governance.source_effective_limits + return { + key: bool(getattr(governance, key)) and source.get(key, True) is True + for key in _LIMITS + } + + +def _principal_ref(principal: object) -> PrincipalRef: + converter = getattr(principal, "to_platform_principal", None) + if callable(converter): + converted = converter() + if isinstance(converted, PrincipalRef): + return converted + return PrincipalRef( + account_id=str(getattr(principal, "account_id", "") or "system"), + membership_id=_optional(getattr(principal, "membership_id", None)), + tenant_id=_tenant(principal), + identity_id=_optional(getattr(principal, "identity_id", None)), + scopes=frozenset(_string_set(getattr(principal, "scopes", ()))), + group_ids=frozenset(_string_set(getattr(principal, "group_ids", ()))), + role_ids=frozenset(_string_set(getattr(principal, "role_ids", ()))), + function_assignment_ids=frozenset( + _string_set(getattr(principal, "function_assignment_ids", ())) + ), + service_account_id=_optional(getattr(principal, "service_account_id", None)), + acting_assignment_id=_optional( + getattr(principal, "acting_assignment_id", None) + ), + ) + + +def _has_scope(principal: object, scope: str) -> bool: + method = getattr(principal, "has", None) + if callable(method): + return bool(method(scope)) + return scope in _string_set(getattr(principal, "scopes", ())) + + +def _tenant(principal: object) -> str: + tenant_id = str(getattr(principal, "tenant_id", "") or "").strip() + if not tenant_id: + raise ReportingGovernanceError( + "Reporting governance requires a tenant-bound principal." + ) + return tenant_id + + +def _string_set(value: object) -> set[str]: + if isinstance(value, (str, bytes)): + return {str(value)} if value else set() + try: + return {str(item) for item in value or () if str(item).strip()} # type: ignore[union-attr] + except TypeError: + return set() + + +def _optional(value: object) -> str | None: + clean = str(value or "").strip() + return clean or None + + +__all__ = [ + "ReportingGovernanceError", + "apply_parent_governance", + "definition_decision", + "governance_payload", + "normalize_definition_governance", + "require_definition_action", + "scope_visible", + "validate_parent_governance", +] + + +__all__ = [ + "ReportingGovernanceError", + "definition_decision", + "governance_payload", + "normalize_definition_governance", + "require_definition_action", + "scope_visible", + "validate_parent_governance", +] diff --git a/src/govoplan_reporting/backend/manifest.py b/src/govoplan_reporting/backend/manifest.py index 2607d58..6ce66e7 100644 --- a/src/govoplan_reporting/backend/manifest.py +++ b/src/govoplan_reporting/backend/manifest.py @@ -7,6 +7,8 @@ from govoplan_core.core.access import ( CAPABILITY_AUTH_PRINCIPAL_RESOLVER, ) from govoplan_core.core.dataflows import CAPABILITY_DATAFLOW_DATASET_OUTPUT +from govoplan_core.core.files import CAPABILITY_FILES_ARTIFACT_STORE +from govoplan_core.core.mail import CAPABILITY_MAIL_NOTIFICATION_DELIVERY from govoplan_core.core.module_guards import ( drop_table_retirement_provider, persistent_table_uninstall_guard, @@ -41,6 +43,8 @@ from govoplan_core.db.base import Base from govoplan_reporting.backend.acl import ReportingScopeAclProvider from govoplan_reporting.backend.contracts import ( CAPABILITY_REPORTING_CHART_RENDERER, + CAPABILITY_REPORTING_PUBLICATION_FILES, + CAPABILITY_REPORTING_PUBLICATION_MAIL, CAPABILITY_REPORTING_REGISTRY, CAPABILITY_REPORTING_RUNNER, CAPABILITY_REPORTING_SCHEDULER, @@ -63,6 +67,10 @@ from govoplan_reporting.backend.operations import ( SqlReportingScheduler, ) from govoplan_reporting.backend.query_engine import DefaultChartRenderer +from govoplan_reporting.backend.publication_targets import ( + FilesReportingPublicationTarget, + MailReportingPublicationTarget, +) from govoplan_reporting.backend.registry import SqlReportingRegistry from govoplan_reporting.backend.search_source import create_reporting_search_source @@ -175,6 +183,14 @@ def _chart_renderer(context: ModuleContext) -> DefaultChartRenderer: return DefaultChartRenderer() +def _files_publication(context: ModuleContext) -> FilesReportingPublicationTarget: + return FilesReportingPublicationTarget(context.registry) + + +def _mail_publication(context: ModuleContext) -> MailReportingPublicationTarget: + return MailReportingPublicationTarget(context.registry) + + def _retention(context: ModuleContext): del context from govoplan_reporting.backend.retention import ReportingRetentionService @@ -246,6 +262,8 @@ manifest = ModuleManifest( optional_capabilities=( CAPABILITY_DATAFLOW_DATASET_OUTPUT, CAPABILITY_POLICY_REPORTING_GOVERNANCE, + CAPABILITY_FILES_ARTIFACT_STORE, + CAPABILITY_MAIL_NOTIFICATION_DELIVERY, ), permissions=PERMISSIONS, role_templates=ROLE_TEMPLATES, @@ -306,6 +324,13 @@ manifest = ModuleManifest( parent_id="reporting.workspace", order=40, ), + ViewSurface( + id="reporting.widget.reports", + module_id=MODULE_ID, + kind="section", + label="Reports dashboard widget", + order=75, + ), ), ), provides_interfaces=( @@ -313,6 +338,12 @@ manifest = ModuleManifest( ModuleInterfaceProvider(name="reporting.runner", version="0.1.0"), ModuleInterfaceProvider(name="reporting.scheduler", version="0.1.0"), ModuleInterfaceProvider(name="reporting.chart_renderer", version="0.1.0"), + ModuleInterfaceProvider( + name=CAPABILITY_REPORTING_PUBLICATION_FILES, version="1.0.0" + ), + ModuleInterfaceProvider( + name=CAPABILITY_REPORTING_PUBLICATION_MAIL, version="1.0.0" + ), ModuleInterfaceProvider(name=CAPABILITY_REPORTING_RETENTION, version="1.0.0"), ), requires_interfaces=( @@ -328,12 +359,26 @@ manifest = ModuleManifest( version_max_exclusive="2.0.0", optional=True, ), + ModuleInterfaceRequirement( + name=CAPABILITY_FILES_ARTIFACT_STORE, + version_min="0.1.14", + version_max_exclusive="0.2.0", + optional=True, + ), + ModuleInterfaceRequirement( + name="mail.notification_delivery", + version_min="0.1.0", + version_max_exclusive="2.0.0", + optional=True, + ), ), capability_factories={ CAPABILITY_REPORTING_REGISTRY: _registry, CAPABILITY_REPORTING_RUNNER: _runner, CAPABILITY_REPORTING_SCHEDULER: _scheduler, CAPABILITY_REPORTING_CHART_RENDERER: _chart_renderer, + CAPABILITY_REPORTING_PUBLICATION_FILES: _files_publication, + CAPABILITY_REPORTING_PUBLICATION_MAIL: _mail_publication, CAPABILITY_REPORTING_RETENTION: _retention, }, capability_documentation={ @@ -357,6 +402,16 @@ manifest = ModuleManifest( summary="Builds provider-neutral chart models with an accessible tabular fallback.", contract_version="0.1.0", ), + CAPABILITY_REPORTING_PUBLICATION_FILES: CapabilityDocumentation( + label="Files report publication", + summary="Stores an immutable authorized report output through Files managed artifact storage.", + contract_version="1.0.0", + ), + CAPABILITY_REPORTING_PUBLICATION_MAIL: CapabilityDocumentation( + label="Mail report publication", + summary="Submits an idempotent report notice through Mail's durable delivery outbox.", + contract_version="1.0.0", + ), CAPABILITY_REPORTING_RETENTION: CapabilityDocumentation( label="Reporting result retention", summary="Minimizes expired provider-report detail while retaining audit hashes and provenance.", @@ -384,6 +439,7 @@ manifest = ModuleManifest( reporting_models.ReportingSavedView, reporting_models.ReportingDefinitionGrant, reporting_models.ReportingExecution, + reporting_models.ReportingDrillContext, reporting_models.ReportingProviderExport, reporting_models.ReportingProviderExecution, reporting_models.ReportingDefinitionRevision, @@ -401,6 +457,7 @@ manifest = ModuleManifest( reporting_models.ReportingDefinitionRevision, reporting_models.ReportingDefinitionGrant, reporting_models.ReportingExecution, + reporting_models.ReportingDrillContext, reporting_models.ReportingProviderExecution, reporting_models.ReportingProviderExport, reporting_models.ReportingSavedView, @@ -429,7 +486,9 @@ manifest = ModuleManifest( "authorized result rows, diagnostics, and output hashes. Safe dimensions, " "aggregations, typed expressions, filters, pivots, saved views, chart models, " "schedules, exports, and publication providers replace unchecked SQL in the " - "presentation layer. Dataflow and module read models remain source owners." + "presentation layer. PostgreSQL executes bounded semantic plans when available. " + "Signed drill contexts reauthorize contributor rows, and Files/Mail publication " + "adapters retain idempotent evidence. Dataflow and module read models remain source owners." ), layer="available", documentation_types=("admin", "user"), @@ -481,9 +540,9 @@ manifest = ModuleManifest( ), known_limits=( "Dataflow is the first live dataset adapter; additional module read models use the provider-neutral contract.", - "Direct browser export supports CSV and JSON; XLSX, PDF, Files, Mail, and DMS delivery require an optional publication provider.", + "Direct browser export supports CSV and JSON. Files supports CSV, JSON, and HTML publication; Mail submits a bounded report notice. XLSX/PDF require a renderer provider.", "Import assessment produces blocking diagnostics but does not execute source SQL or automatically activate generated definitions.", - "The initial chart provider emits a renderer-neutral model and accessible table; richer visual renderers remain replaceable adapters.", + "The built-in chart catalogue covers bounded bar, column, line, area, pie, donut, and metric views; specialized visual renderers remain replaceable adapters.", ), owned_concepts=( "analytical dataset binding", diff --git a/src/govoplan_reporting/backend/migrations/versions/c8d5e2f6a9b3_reporting_drill_contexts.py b/src/govoplan_reporting/backend/migrations/versions/c8d5e2f6a9b3_reporting_drill_contexts.py new file mode 100644 index 0000000..90dcf84 --- /dev/null +++ b/src/govoplan_reporting/backend/migrations/versions/c8d5e2f6a9b3_reporting_drill_contexts.py @@ -0,0 +1,71 @@ +"""Add authorization-bound Reporting drill contexts. + +Revision ID: c8d5e2f6a9b3 +Revises: b7c4e1a9d2f6 +""" + +from __future__ import annotations + +from alembic import op +import sqlalchemy as sa + + +revision = "c8d5e2f6a9b3" +down_revision = "b7c4e1a9d2f6" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.create_table( + "reporting_drill_contexts", + sa.Column("id", sa.String(length=36), nullable=False), + sa.Column("tenant_id", sa.String(length=36), nullable=False), + sa.Column("drill_context_id", sa.String(length=36), nullable=False), + sa.Column("execution_id", sa.String(length=36), nullable=False), + sa.Column("token_sha256", sa.String(length=64), nullable=False), + sa.Column("context_sha256", sa.String(length=64), nullable=False), + sa.Column("actor_id", sa.String(length=255), nullable=False), + sa.Column("dimension_path", sa.JSON(), nullable=False), + sa.Column("source_fingerprints", sa.JSON(), nullable=False), + sa.Column("policy_provenance", sa.JSON(), nullable=False), + sa.Column("expires_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("last_accessed_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), + sa.PrimaryKeyConstraint("id", name=op.f("pk_reporting_drill_contexts")), + sa.UniqueConstraint( + "tenant_id", + "drill_context_id", + name="uq_reporting_drill_context", + ), + ) + for column in ( + "tenant_id", + "drill_context_id", + "execution_id", + "actor_id", + "expires_at", + ): + op.create_index( + op.f(f"ix_reporting_drill_contexts_{column}"), + "reporting_drill_contexts", + [column], + unique=False, + ) + op.create_index( + "ix_reporting_drill_context_expiry", + "reporting_drill_contexts", + ["tenant_id", "expires_at"], + unique=False, + ) + op.create_index( + "ix_reporting_drill_context_execution", + "reporting_drill_contexts", + ["tenant_id", "execution_id"], + unique=False, + ) + + +def downgrade() -> None: + op.drop_table("reporting_drill_contexts") diff --git a/src/govoplan_reporting/backend/operations.py b/src/govoplan_reporting/backend/operations.py index 1930ea7..bd3ac9f 100644 --- a/src/govoplan_reporting/backend/operations.py +++ b/src/govoplan_reporting/backend/operations.py @@ -396,7 +396,12 @@ def publish_execution( options: Mapping[str, object], ) -> dict[str, object]: _require_scope(principal, PUBLISH_SCOPE) - execution_payload = get_execution(session, principal, execution_id=execution_id) + execution_payload = get_execution( + session, + principal, + execution_id=execution_id, + registry=registry, + ) if execution_payload is None: raise LookupError("Reporting execution not found.") if execution_payload["status"] != "succeeded": @@ -497,14 +502,54 @@ def publish_execution( return _publication_payload(publication) +def list_publications( + session: Session, + principal: object, + *, + execution_id: str | None = None, + limit: int = 100, + registry: object | None = None, +) -> tuple[dict[str, object], ...]: + _require_scope(principal, PUBLISH_SCOPE) + statement = session.query(ReportingPublication).filter( + ReportingPublication.tenant_id == _tenant(principal) + ) + if execution_id: + if ( + get_execution( + session, + principal, + execution_id=execution_id, + registry=registry, + ) + is None + ): + return () + statement = statement.filter( + ReportingPublication.execution_id == execution_id + ) + rows = ( + statement.order_by(ReportingPublication.created_at.desc()) + .limit(max(1, min(limit, 200))) + .all() + ) + return tuple(_publication_payload(row) for row in rows) + + def export_execution( session: Session, principal: object, *, execution_id: str, format: str, + registry: object | None = None, ) -> tuple[bytes, str, str]: - payload = get_execution(session, principal, execution_id=execution_id) + payload = get_execution( + session, + principal, + execution_id=execution_id, + registry=registry, + ) if payload is None: raise LookupError("Reporting execution not found.") if payload["status"] != "succeeded": @@ -850,6 +895,7 @@ __all__ = [ "dispatch_due_schedules", "export_execution", "list_import_assessments", + "list_publications", "list_saved_views", "list_schedules", "publish_execution", diff --git a/src/govoplan_reporting/backend/postgres_planner.py b/src/govoplan_reporting/backend/postgres_planner.py new file mode 100644 index 0000000..4376f0f --- /dev/null +++ b/src/govoplan_reporting/backend/postgres_planner.py @@ -0,0 +1,509 @@ +from __future__ import annotations + +from collections.abc import Mapping, Sequence +from datetime import date, datetime +from decimal import Decimal +import json +import re +from typing import Any + +from sqlalchemy import text +from sqlalchemy.orm import Session + +from govoplan_reporting.backend.query_engine import ( + QueryResult, + ReportingQueryError, + infer_query_schema, +) +from govoplan_reporting.backend.schemas import ( + DatasetDefinition, + DimensionDefinition, + FilterClause, + MeasureDefinition, + ReportQuery, + SemanticModelDefinition, + TypedExpression, +) + + +POSTGRES_PLANNER_VERSION = "reporting-postgresql-v1" +_IDENTIFIER = re.compile(r"^[a-z0-9._-]{1,120}$") + + +class PostgresPlanningError(ReportingQueryError): + pass + + +def execute_postgres_query( + session: Session, + *, + rows: Sequence[Mapping[str, object]], + dataset: DatasetDefinition, + semantic_model: SemanticModelDefinition, + query: ReportQuery, +) -> QueryResult | None: + """Execute a bounded semantic plan in PostgreSQL, or return None for fallback.""" + + if session.bind is None or session.bind.dialect.name != "postgresql": + return None + if query.mode == "pivot": + return None + plan = compile_postgres_query(dataset, semantic_model, query) + parameters = { + **plan.parameters, + "rows_json": json.dumps( + [_json_value(dict(item)) for item in rows], + ensure_ascii=True, + separators=(",", ":"), + sort_keys=True, + ), + "result_limit": query.limit, + "result_offset": query.offset, + } + result = session.execute(text(plan.sql), parameters).mappings().all() + total_rows = int(result[0]["__reporting_total"]) if result else 0 + output = tuple( + { + str(key): _json_value(value) + for key, value in item.items() + if key != "__reporting_total" + } + for item in result + ) + return QueryResult( + rows=output, + total_rows=total_rows, + schema=infer_query_schema(output), + truncated=query.offset + len(output) < total_rows, + diagnostics=( + { + "severity": "info", + "code": "postgresql_semantic_plan", + "message": "Filters, grouping, measures, ordering, and bounds were executed by the PostgreSQL Reporting planner.", + "planner_version": POSTGRES_PLANNER_VERSION, + }, + ), + ) + + +class CompiledPostgresPlan: + __slots__ = ("sql", "parameters") + + def __init__(self, sql: str, parameters: Mapping[str, object]) -> None: + self.sql = sql + self.parameters = dict(parameters) + + +def compile_postgres_query( + dataset: DatasetDefinition, + semantic_model: SemanticModelDefinition, + query: ReportQuery, +) -> CompiledPostgresPlan: + dimensions = {item.key: item for item in semantic_model.dimensions} + measures = {item.key: item for item in semantic_model.measures} + selected_dimensions = tuple(query.dimensions or semantic_model.default_dimensions) + selected_measures = tuple(query.measures or semantic_model.default_measures) + _known(selected_dimensions, dimensions, "dimensions") + _known(selected_measures, measures, "measures") + _known( + tuple(item.dimension for item in query.filters), dimensions, "filter dimensions" + ) + selected_keys = set(selected_dimensions) + if query.mode != "detail": + selected_keys.update(selected_measures) + _known( + tuple(item.key for item in query.sort), + {key: True for key in selected_keys}, + "sort fields", + ) + + parameters: dict[str, object] = {} + source = ( + "WITH source AS (" + "SELECT value AS source_row " + "FROM jsonb_array_elements(CAST(:rows_json AS jsonb)) AS source_items(value)" + ")" + ) + where = _filter_sql(query.filters, dimensions, parameters) + if query.mode == "detail": + fields = selected_dimensions + if not fields: + if not dataset.fields: + raise PostgresPlanningError( + "PostgreSQL detail planning requires selected dimensions or a pinned dataset schema." + ) + field_types = {item.name: item.type for item in dataset.fields} + projections = [ + f"{_source_value(item.name, item.type, parameters, f'detail_{index}')} AS {_quote(item.name)}" + for index, item in enumerate(dataset.fields) + ] + selected_keys = set(field_types) + else: + projections = [ + f"{_dimension_value(dimensions[key], parameters, f'detail_{index}')} AS {_quote(key)}" + for index, key in enumerate(fields) + ] + body = "SELECT " + ", ".join(projections) + " FROM source" + where + else: + dimension_projections = [ + ( + key, + _dimension_value(dimensions[key], parameters, f"dimension_{index}"), + ) + for index, key in enumerate(selected_dimensions) + ] + selected_base_keys = [ + key + for key in selected_measures + if measures[key].aggregation != "calculated" + ] + calculated = [ + measures[key] + for key in selected_measures + if measures[key].aggregation == "calculated" + ] + dependency_keys = list( + dict.fromkeys( + dependency + for item in calculated + for dependency in _calculated_dependencies( + item.expression, measures, stack=(item.key,) + ) + ) + ) + base_measure_keys = list(dict.fromkeys((*selected_base_keys, *dependency_keys))) + base_measures = [measures[key] for key in base_measure_keys] + grouped_select = [ + f"{expression} AS {_quote(key)}" + for key, expression in dimension_projections + ] + [ + f"{_aggregate_sql(item, parameters, index)} AS {_quote(item.key)}" + for index, item in enumerate(base_measures) + ] + if not grouped_select: + raise PostgresPlanningError( + "Summary queries require at least one dimension or measure." + ) + grouped = "SELECT " + ", ".join(grouped_select) + " FROM source" + where + if dimension_projections: + grouped += " GROUP BY " + ", ".join( + expression for _key, expression in dimension_projections + ) + if calculated: + outer = [_quote(key) for key in selected_dimensions] + [ + _quote(key) for key in selected_base_keys + ] + outer.extend( + f"{_calculated_sql(item.expression, parameters, f'calculated_{index}', measures=measures, stack=(item.key,))} AS {_quote(item.key)}" + for index, item in enumerate(calculated) + ) + body = "SELECT " + ", ".join(outer) + f" FROM ({grouped}) AS grouped" + else: + body = grouped + order = "" + if query.sort: + order = " ORDER BY " + ", ".join( + f"{_quote(item.key)} {item.direction.upper()} NULLS LAST" + for item in query.sort + ) + sql = ( + source + + " SELECT planned.*, COUNT(*) OVER() AS __reporting_total FROM (" + + body + + ") AS planned" + + order + + " LIMIT :result_limit OFFSET :result_offset" + ) + return CompiledPostgresPlan(sql, parameters) + + +def _filter_sql( + filters: Sequence[FilterClause], + dimensions: Mapping[str, DimensionDefinition], + parameters: dict[str, object], +) -> str: + clauses: list[str] = [] + for index, clause in enumerate(filters): + value = _dimension_value( + dimensions[clause.dimension], parameters, f"filter_field_{index}" + ) + prefix = f"filter_{index}" + if clause.operator == "is_null": + clauses.append(f"{value} IS NULL") + continue + if clause.operator == "not_null": + clauses.append(f"{value} IS NOT NULL") + continue + if clause.operator in {"in", "not_in"}: + if not isinstance(clause.value, (list, tuple)): + raise PostgresPlanningError("Set filters require a list value.") + if not clause.value or len(clause.value) > 500: + raise PostgresPlanningError( + "Set filters require between 1 and 500 values." + ) + names: list[str] = [] + for item_index, item in enumerate(clause.value): + name = f"{prefix}_{item_index}" + parameters[name] = item + names.append(f":{name}") + operator = "NOT IN" if clause.operator == "not_in" else "IN" + clauses.append(f"{value} {operator} ({', '.join(names)})") + continue + if clause.operator == "between": + if not isinstance(clause.value, (list, tuple)) or len(clause.value) != 2: + raise PostgresPlanningError("Between filters require two values.") + parameters[f"{prefix}_low"] = clause.value[0] + parameters[f"{prefix}_high"] = clause.value[1] + clauses.append(f"{value} BETWEEN :{prefix}_low AND :{prefix}_high") + continue + parameters[prefix] = clause.value + if clause.operator == "contains": + parameters[prefix] = f"%{_like(str(clause.value or ''))}%" + clauses.append( + f"LOWER(CAST({value} AS text)) LIKE LOWER(:{prefix}) ESCAPE '\\'" + ) + elif clause.operator == "starts_with": + parameters[prefix] = f"{_like(str(clause.value or ''))}%" + clauses.append( + f"LOWER(CAST({value} AS text)) LIKE LOWER(:{prefix}) ESCAPE '\\'" + ) + else: + operator = { + "eq": "=", + "ne": "<>", + "gt": ">", + "gte": ">=", + "lt": "<", + "lte": "<=", + }.get(clause.operator) + if operator is None: + raise PostgresPlanningError( + f"Unsupported PostgreSQL filter operator: {clause.operator}." + ) + clauses.append(f"{value} {operator} :{prefix}") + return " WHERE " + " AND ".join(clauses) if clauses else "" + + +def _aggregate_sql( + measure: MeasureDefinition, + parameters: dict[str, object], + index: int, +) -> str: + if measure.aggregation == "count" and measure.field is None: + return "COUNT(*)" + field = _source_value( + measure.field or "", + "number" if measure.aggregation in {"sum", "average"} else "string", + parameters, + f"measure_{index}", + ) + if measure.aggregation == "count": + return f"COUNT({field})" + if measure.aggregation == "count_distinct": + return f"COUNT(DISTINCT {field})" + function = { + "sum": "SUM", + "average": "AVG", + "minimum": "MIN", + "maximum": "MAX", + }.get(measure.aggregation) + if function is None: + raise PostgresPlanningError( + f"Unsupported PostgreSQL aggregation: {measure.aggregation}." + ) + return f"{function}({field})" + + +def _calculated_sql( + expression: TypedExpression | None, + parameters: dict[str, object], + prefix: str, + *, + measures: Mapping[str, MeasureDefinition], + stack: tuple[str, ...], +) -> str: + if expression is None: + return "NULL" + if expression.op == "literal": + parameters[prefix] = expression.value + return f":{prefix}" + if expression.op == "measure": + reference = expression.ref or "" + target = measures.get(reference) + if target is None: + raise PostgresPlanningError( + f"Calculated measure references unknown measure: {reference}." + ) + if target.aggregation != "calculated": + return _quote(reference) + if reference in stack: + raise PostgresPlanningError( + "Calculated measure dependency cycle: " + + " -> ".join((*stack, reference)) + ) + return _calculated_sql( + target.expression, + parameters, + prefix + "_" + reference, + measures=measures, + stack=(*stack, reference), + ) + if expression.op == "field": + raise PostgresPlanningError( + "Calculated aggregate measures may reference measures, not source fields." + ) + values = [ + _calculated_sql( + item, + parameters, + f"{prefix}_{index}", + measures=measures, + stack=stack, + ) + for index, item in enumerate(expression.args) + ] + if expression.op in {"add", "multiply", "and", "or"}: + operator = {"add": "+", "multiply": "*", "and": "AND", "or": "OR"}[ + expression.op + ] + return "(" + f" {operator} ".join(values) + ")" + if expression.op in {"subtract", "divide", "eq", "ne", "gt", "gte", "lt", "lte"}: + if len(values) != 2: + raise PostgresPlanningError( + f"Expression {expression.op} requires exactly two arguments." + ) + operator = { + "subtract": "-", + "divide": "/", + "eq": "=", + "ne": "<>", + "gt": ">", + "gte": ">=", + "lt": "<", + "lte": "<=", + }[expression.op] + right = f"NULLIF({values[1]}, 0)" if expression.op == "divide" else values[1] + return f"({values[0]} {operator} {right})" + if expression.op == "not": + if len(values) != 1: + raise PostgresPlanningError("Expression not requires one argument.") + return f"(NOT {values[0]})" + if expression.op == "coalesce": + return "COALESCE(" + ", ".join(values) + ")" + if expression.op == "case": + if len(values) < 3 or len(values) % 2 == 0: + raise PostgresPlanningError( + "Case expressions require condition/value pairs and a default." + ) + branches = " ".join( + f"WHEN {values[index]} THEN {values[index + 1]}" + for index in range(0, len(values) - 1, 2) + ) + return f"(CASE {branches} ELSE {values[-1]} END)" + raise PostgresPlanningError( + f"Unsupported PostgreSQL expression operator: {expression.op}." + ) + + +def _calculated_dependencies( + expression: TypedExpression | None, + measures: Mapping[str, MeasureDefinition], + *, + stack: tuple[str, ...], +) -> tuple[str, ...]: + if expression is None: + return () + if expression.op == "measure": + reference = expression.ref or "" + target = measures.get(reference) + if target is None: + raise PostgresPlanningError( + f"Calculated measure references unknown measure: {reference}." + ) + if target.aggregation != "calculated": + return (reference,) + if reference in stack: + raise PostgresPlanningError( + "Calculated measure dependency cycle: " + + " -> ".join((*stack, reference)) + ) + return _calculated_dependencies( + target.expression, + measures, + stack=(*stack, reference), + ) + dependencies: list[str] = [] + for item in expression.args: + dependencies.extend(_calculated_dependencies(item, measures, stack=stack)) + return tuple(dict.fromkeys(dependencies)) + + +def _dimension_value( + dimension: DimensionDefinition, + parameters: dict[str, object], + prefix: str, +) -> str: + return _source_value(dimension.field, dimension.type, parameters, prefix) + + +def _source_value( + field: str, + field_type: str, + parameters: dict[str, object], + prefix: str, +) -> str: + parameters[prefix] = field + raw = f"source_row ->> :{prefix}" + if field_type == "integer": + return f"NULLIF({raw}, '')::bigint" + if field_type == "number": + return f"NULLIF({raw}, '')::numeric" + if field_type == "boolean": + return f"NULLIF({raw}, '')::boolean" + if field_type == "date": + return f"NULLIF({raw}, '')::date" + if field_type == "datetime": + return f"NULLIF({raw}, '')::timestamptz" + if field_type == "json": + return f"source_row -> :{prefix}" + return raw + + +def _known(keys: Sequence[str], available: Mapping[str, object], label: str) -> None: + unknown = set(keys) - set(available) + if unknown: + raise PostgresPlanningError( + f"Report query references unknown {label}: " + ", ".join(sorted(unknown)) + ) + + +def _quote(value: str) -> str: + if not _IDENTIFIER.fullmatch(value): + raise PostgresPlanningError(f"Unsafe Reporting identifier: {value!r}.") + return '"' + value.replace('"', '""') + '"' + + +def _like(value: str) -> str: + return value.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") + + +def _json_value(value: object) -> Any: + if isinstance(value, Decimal): + integral = value.to_integral_value() + return int(integral) if value == integral else float(value) + if isinstance(value, (datetime, date)): + return value.isoformat() + 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 + + +__all__ = [ + "POSTGRES_PLANNER_VERSION", + "CompiledPostgresPlan", + "PostgresPlanningError", + "compile_postgres_query", + "execute_postgres_query", +] diff --git a/src/govoplan_reporting/backend/publication_targets.py b/src/govoplan_reporting/backend/publication_targets.py new file mode 100644 index 0000000..1962fcf --- /dev/null +++ b/src/govoplan_reporting/backend/publication_targets.py @@ -0,0 +1,303 @@ +from __future__ import annotations + +from collections.abc import Mapping, Sequence +import csv +from html import escape +from io import StringIO +import json +import re + +from govoplan_core.core.files import ( + CAPABILITY_FILES_ARTIFACT_STORE, + ManagedArtifactStore, + ManagedArtifactWriteRequest, +) +from govoplan_core.core.mail import ( + CAPABILITY_MAIL_NOTIFICATION_DELIVERY, + NotificationMailDeliveryProvider, + NotificationMailDeliveryRequest, +) +from govoplan_reporting.backend.contracts import ( + CAPABILITY_REPORTING_PUBLICATION_FILES, + CAPABILITY_REPORTING_PUBLICATION_MAIL, + ReportingPublicationPayload, + capability, +) + + +class FilesReportingPublicationTarget: + def __init__(self, registry: object | None) -> None: + self.registry = registry + + def publish_report( + self, + session: object, + principal: object, + *, + payload: ReportingPublicationPayload, + ) -> Mapping[str, object]: + provider = capability(self.registry, CAPABILITY_FILES_ARTIFACT_STORE) + if not isinstance(provider, ManagedArtifactStore): + raise RuntimeError( + "Files publication requires the enabled files.artifact_store capability." + ) + content, content_type, extension = _serialize(payload) + filename = _filename(payload, extension) + folder = str(payload.target_ref or "Generated/Reports").strip() + stored = provider.store_artifact( + session, + principal, + request=ManagedArtifactWriteRequest( + filename=filename, + payload=content, + content_type=content_type, + folder=folder, + description=( + f"Reporting publication for {payload.report_id} revision " + f"{payload.report_revision}." + ), + idempotency_key=f"reporting:{payload.publication_id}", + metadata={ + "producer_module": "reporting", + "publication_id": payload.publication_id, + "execution_id": payload.execution_id, + "report_id": payload.report_id, + "report_revision": payload.report_revision, + "output_hash": payload.output_hash, + }, + ), + ) + return { + "provider": CAPABILITY_FILES_ARTIFACT_STORE, + "status": "stored", + "file_asset_id": stored.file_asset_id, + "file_version_id": stored.file_version_id, + "filename": stored.filename, + "display_path": stored.display_path, + "sha256": stored.sha256, + "size_bytes": stored.size_bytes, + "output_hash": payload.output_hash, + } + + +class MailReportingPublicationTarget: + def __init__(self, registry: object | None) -> None: + self.registry = registry + + def publish_report( + self, + session: object, + principal: object, + *, + payload: ReportingPublicationPayload, + ) -> Mapping[str, object]: + provider = capability(self.registry, CAPABILITY_MAIL_NOTIFICATION_DELIVERY) + if not isinstance(provider, NotificationMailDeliveryProvider): + raise RuntimeError( + "Mail publication requires the enabled mail.notificationDelivery capability." + ) + recipient = str(payload.target_ref or "").strip() + if not recipient: + raise ValueError("Mail publication requires a recipient address.") + options = dict(payload.options) + profile_id = _required_option(options, "mail_profile_id", "Mail profile") + from_address = _required_option(options, "from_address", "Sender address") + subject = str( + options.get("subject") + or f"Report {payload.report_id} revision {payload.report_revision}" + ).strip() + action_url = str(options.get("action_url") or "").strip() or None + preview = _text_preview(payload.rows, payload.schema) + result = provider.submit_notification_mail( + session, + NotificationMailDeliveryRequest( + tenant_id=payload.tenant_id, + notification_id=f"reporting-publication:{payload.publication_id}", + recipient=recipient, + subject=subject, + body_text=( + f"Report: {payload.report_id}\n" + f"Revision: {payload.report_revision}\n" + f"Rows: {len(payload.rows)}\n" + f"Output hash: {payload.output_hash}\n\n" + f"{preview}" + ), + action_url=action_url, + mail_profile_id=profile_id, + from_address=from_address, + smtp_server_id=_optional(options.get("smtp_server_id")), + smtp_credential_id=_optional(options.get("smtp_credential_id")), + metadata={ + "producer_module": "reporting", + "publication_id": payload.publication_id, + "execution_id": payload.execution_id, + "report_id": payload.report_id, + "report_revision": payload.report_revision, + "output_hash": payload.output_hash, + }, + ), + ) + status = str(result.get("status") or "").casefold() + if status not in {"accepted", "queued", "submitted", "succeeded"}: + raise RuntimeError( + str(result.get("error") or "Mail did not accept the report publication.") + ) + return { + **dict(result), + "publication_id": payload.publication_id, + "recipient": recipient, + "output_hash": payload.output_hash, + } + + +def publication_target_catalog(registry: object | None) -> tuple[dict[str, object], ...]: + files_available = isinstance( + capability(registry, CAPABILITY_FILES_ARTIFACT_STORE), ManagedArtifactStore + ) + mail_available = isinstance( + capability(registry, CAPABILITY_MAIL_NOTIFICATION_DELIVERY), + NotificationMailDeliveryProvider, + ) + return ( + { + "capability": CAPABILITY_REPORTING_PUBLICATION_FILES, + "label": "Files", + "available": files_available, + "reason": None + if files_available + else "Enable Files with managed artifact storage to publish durable report files.", + "formats": ["csv", "json", "html"], + "target_label": "Folder", + "target_required": False, + "required_options": [], + }, + { + "capability": CAPABILITY_REPORTING_PUBLICATION_MAIL, + "label": "Mail", + "available": mail_available, + "reason": None + if mail_available + else "Enable Mail and configure its notification-delivery capability to publish report notices.", + "formats": ["html"], + "target_label": "Recipient", + "target_required": True, + "required_options": ["mail_profile_id", "from_address"], + }, + ) + + +def _serialize(payload: ReportingPublicationPayload) -> tuple[bytes, str, str]: + if payload.format == "json": + content = json.dumps( + { + "report_id": payload.report_id, + "report_revision": payload.report_revision, + "execution_id": payload.execution_id, + "output_hash": payload.output_hash, + "schema": list(payload.schema), + "rows": list(payload.rows), + }, + ensure_ascii=False, + indent=2, + default=str, + ).encode("utf-8") + return content, "application/json", "json" + if payload.format == "csv": + fields = _fields(payload.rows, payload.schema) + stream = StringIO(newline="") + writer = csv.DictWriter(stream, fieldnames=fields, extrasaction="ignore") + writer.writeheader() + for row in payload.rows: + writer.writerow({key: _safe_csv(row.get(key)) for key in fields}) + return ( + stream.getvalue().encode("utf-8-sig"), + "text/csv; charset=utf-8", + "csv", + ) + if payload.format == "html": + fields = _fields(payload.rows, payload.schema) + headers = "".join(f"{escape(key)}" for key in fields) + body = "".join( + "" + + "".join( + f"{escape(_display(row.get(key)))}" for key in fields + ) + + "" + for row in payload.rows + ) + content = ( + "" + + escape(payload.report_id) + + "

" + + escape(payload.report_id) + + f"

Revision {payload.report_revision}; output {escape(payload.output_hash)}

" + + f"{headers}{body}
" + + "" + ) + return content.encode("utf-8"), "text/html; charset=utf-8", "html" + raise ValueError( + "This publication target supports CSV, JSON, and accessible HTML. " + "XLSX and PDF require a renderer provider." + ) + + +def _filename(payload: ReportingPublicationPayload, extension: str) -> str: + configured = str(payload.options.get("filename") or "").strip() + stem = configured.rsplit(".", 1)[0] if configured else payload.report_id + safe = re.sub(r"[^A-Za-z0-9._-]+", "-", stem).strip(".-") or "report" + return f"{safe}-r{payload.report_revision}.{extension}" + + +def _fields( + rows: Sequence[Mapping[str, object]], schema: Sequence[Mapping[str, object]] +) -> list[str]: + fields = [str(item.get("name")) for item in schema if item.get("name")] + if fields: + return fields + return list(dict.fromkeys(str(key) for row in rows for key in row)) + + +def _safe_csv(value: object) -> object: + if isinstance(value, (dict, list, tuple)): + value = json.dumps(value, ensure_ascii=False, separators=(",", ":")) + if isinstance(value, str) and value.startswith(("=", "+", "-", "@")): + return "'" + value + return value + + +def _display(value: object) -> str: + if value is None: + return "" + if isinstance(value, (dict, list, tuple)): + return json.dumps(value, ensure_ascii=False, separators=(",", ":")) + return str(value) + + +def _text_preview( + rows: Sequence[Mapping[str, object]], schema: Sequence[Mapping[str, object]] +) -> str: + fields = _fields(rows, schema)[:8] + lines = [" | ".join(fields)] + lines.extend(" | ".join(_display(row.get(key)) for key in fields) for row in rows[:10]) + if len(rows) > 10: + lines.append(f"... {len(rows) - 10} more rows") + return "\n".join(lines) + + +def _required_option(options: Mapping[str, object], key: str, label: str) -> str: + value = str(options.get(key) or "").strip() + if not value: + raise ValueError(f"{label} is required for Mail publication.") + return value + + +def _optional(value: object) -> str | None: + clean = str(value or "").strip() + return clean or None + + +__all__ = [ + "FilesReportingPublicationTarget", + "MailReportingPublicationTarget", + "publication_target_catalog", +] diff --git a/src/govoplan_reporting/backend/query_engine.py b/src/govoplan_reporting/backend/query_engine.py index 69b1bc3..bb3ecb5 100644 --- a/src/govoplan_reporting/backend/query_engine.py +++ b/src/govoplan_reporting/backend/query_engine.py @@ -86,7 +86,7 @@ def execute_semantic_query( return QueryResult( rows=tuple(selected), total_rows=total, - schema=_infer_schema(selected or sorted_rows[:1]), + schema=infer_query_schema(selected or sorted_rows[:1]), truncated=query.offset + len(selected) < total, ) @@ -376,7 +376,7 @@ def _sort_rows( return result -def _infer_schema(rows: Sequence[Mapping[str, object]]) -> tuple[dict[str, Any], ...]: +def infer_query_schema(rows: Sequence[Mapping[str, object]]) -> tuple[dict[str, Any], ...]: names = tuple(dict.fromkeys(str(key) for row in rows for key in row)) return tuple( { @@ -491,4 +491,5 @@ __all__ = [ "QueryResult", "ReportingQueryError", "execute_semantic_query", + "infer_query_schema", ] diff --git a/src/govoplan_reporting/backend/router.py b/src/govoplan_reporting/backend/router.py index 03f55dd..e557000 100644 --- a/src/govoplan_reporting/backend/router.py +++ b/src/govoplan_reporting/backend/router.py @@ -18,6 +18,11 @@ from govoplan_reporting.backend.definitions import ( list_definitions, update_definition, ) +from govoplan_reporting.backend.drilldown import ( + ReportingDrillError, + create_drill_context, + resolve_drill_context, +) from govoplan_reporting.backend.execution import ( QUALITY_SCOPE, RUN_SCOPE, @@ -38,6 +43,7 @@ from govoplan_reporting.backend.operations import ( dispatch_due_schedules, export_execution, list_import_assessments, + list_publications, list_saved_views, list_schedules, publish_execution, @@ -53,10 +59,12 @@ from govoplan_reporting.backend.provider_reports import ( list_provider_reports, provider_parameter_options, ) +from govoplan_reporting.backend.publication_targets import publication_target_catalog from govoplan_reporting.backend.query_engine import ReportingQueryError from govoplan_reporting.backend.schemas import ( DefinitionUpdateRequest, DefinitionWriteRequest, + DrillContextCreateRequest, ImportAssessmentRequest, PublicationRequest, ProviderReportExecutionRequest, @@ -432,7 +440,13 @@ def create_router(registry: object | None) -> APIRouter: _require(principal, RUN_SCOPE) return { "executions": list( - list_executions(session, principal, report_id=report_id, limit=limit) + list_executions( + session, + principal, + report_id=report_id, + limit=limit, + registry=registry, + ) ) } @@ -443,11 +457,69 @@ def create_router(registry: object | None) -> APIRouter: principal: ApiPrincipal = Depends(get_api_principal), ) -> dict[str, object]: _require(principal, RUN_SCOPE) - result = get_execution(session, principal, execution_id=execution_id) + result = get_execution( + session, + principal, + execution_id=execution_id, + registry=registry, + ) if result is None: raise HTTPException(status_code=404, detail="Reporting execution not found") return result + @router.post("/executions/{execution_id}/drill-contexts", status_code=201) + def api_create_drill_context( + execution_id: str, + payload: DrillContextCreateRequest, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), + ) -> dict[str, object]: + _require(principal, RUN_SCOPE) + try: + result = create_drill_context( + session, + principal, + registry=registry, + execution_id=execution_id, + aggregate_row=payload.aggregate_row, + limit=payload.limit, + ) + session.commit() + except ( + ReportingDrillError, + ReportingExecutionError, + PermissionError, + LookupError, + ) as exc: + session.rollback() + raise _error(exc) from exc + return result + + @router.get("/drill-contexts/{token}") + def api_resolve_drill_context( + token: str, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), + ) -> dict[str, object]: + _require(principal, RUN_SCOPE) + try: + result = resolve_drill_context( + session, + principal, + registry=registry, + token=token, + ) + session.commit() + except ( + ReportingDrillError, + ReportingExecutionError, + PermissionError, + LookupError, + ) as exc: + session.rollback() + raise _error(exc) from exc + return result + @router.get("/executions/{execution_id}/export") def api_export_execution( execution_id: str, @@ -462,6 +534,7 @@ def create_router(registry: object | None) -> APIRouter: principal, execution_id=execution_id, format=format, + registry=registry, ) except (ReportingOperationError, LookupError) as exc: raise _error(exc) from exc @@ -493,6 +566,33 @@ def create_router(registry: object | None) -> APIRouter: raise _error(exc) from exc return result + @router.get("/publication-targets") + def api_publication_targets( + principal: ApiPrincipal = Depends(get_api_principal), + ) -> dict[str, object]: + _require(principal, PUBLISH_SCOPE) + return {"targets": list(publication_target_catalog(registry))} + + @router.get("/publications") + def api_list_publications( + execution_id: str | None = None, + limit: int = Query(default=100, ge=1, le=200), + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), + ) -> dict[str, object]: + _require(principal, PUBLISH_SCOPE) + return { + "publications": list( + list_publications( + session, + principal, + execution_id=execution_id, + limit=limit, + registry=registry, + ) + ) + } + @router.get("/reports/{report_id}/saved-views") def api_list_saved_views( report_id: str, diff --git a/src/govoplan_reporting/backend/schemas.py b/src/govoplan_reporting/backend/schemas.py index 237e686..3666e42 100644 --- a/src/govoplan_reporting/backend/schemas.py +++ b/src/govoplan_reporting/backend/schemas.py @@ -58,6 +58,43 @@ class FreshnessPolicy(BaseModel): require_source_fingerprints: bool = True +class DefinitionGovernance(BaseModel): + """Versioned scope and restrictive inheritance metadata for a definition.""" + + model_config = ConfigDict(extra="forbid") + + scope_type: Literal["system", "tenant", "group", "user"] = "tenant" + scope_id: str | None = Field(default=None, max_length=255) + inherit_to_lower_scopes: bool = False + allow_run: bool = True + allow_reuse: bool = False + allow_automation: bool = False + source_scope: dict[str, Any] | None = None + source_effective_limits: dict[str, bool] = Field(default_factory=dict) + derivation_provenance: dict[str, Any] = Field(default_factory=dict) + + @model_validator(mode="after") + def validate_scope(self) -> "DefinitionGovernance": + if self.scope_type == "system": + if self.scope_id: + raise ValueError("System Reporting definitions do not carry a scope ID.") + elif self.scope_type in {"group", "user"} and not str(self.scope_id or "").strip(): + raise ValueError( + f"{self.scope_type.capitalize()} Reporting definitions require a scope ID." + ) + unknown = set(self.source_effective_limits) - { + "inherit_to_lower_scopes", + "allow_run", + "allow_reuse", + "allow_automation", + } + if unknown: + raise ValueError( + "Unknown inherited Reporting limits: " + ", ".join(sorted(unknown)) + ) + return self + + class DatasetDefinition(BaseModel): model_config = ConfigDict(extra="forbid", populate_by_name=True) @@ -87,6 +124,7 @@ class DatasetDefinition(BaseModel): default_factory=list, max_length=200, ) + governance: DefinitionGovernance = Field(default_factory=DefinitionGovernance) @model_validator(mode="after") def validate_source_pin(self) -> "DatasetDefinition": @@ -203,6 +241,7 @@ class SemanticModelDefinition(BaseModel): default_dimensions: list[str] = Field(default_factory=list, max_length=50) default_measures: list[str] = Field(default_factory=list, max_length=50) metadata: dict[str, Any] = Field(default_factory=dict) + governance: DefinitionGovernance = Field(default_factory=DefinitionGovernance) @model_validator(mode="after") def validate_semantics(self) -> "SemanticModelDefinition": @@ -307,6 +346,7 @@ class VisualizationDefinition(BaseModel): "area", "column", "pie", + "donut", "metric", ] = "table" category_dimension: str | None = Field(default=None, max_length=120) @@ -333,6 +373,7 @@ class ReportDefinition(BaseModel): default_factory=list, max_length=200, ) + governance: DefinitionGovernance = Field(default_factory=DefinitionGovernance) class QualityAssertion(BaseModel): @@ -359,6 +400,7 @@ class QualityPlanDefinition(BaseModel): dataset_revision: int = Field(ge=1) assertions: list[QualityAssertion] = Field(min_length=1, max_length=200) block_report_execution: bool = True + governance: DefinitionGovernance = Field(default_factory=DefinitionGovernance) DEFINITION_PAYLOAD_TYPES = { @@ -478,6 +520,13 @@ class PublicationRequest(BaseModel): options: dict[str, Any] = Field(default_factory=dict) +class DrillContextCreateRequest(BaseModel): + model_config = ConfigDict(extra="forbid") + + aggregate_row: dict[str, Any] = Field(max_length=500) + limit: int = Field(default=200, ge=1, le=500) + + class QualityRunRequest(BaseModel): model_config = ConfigDict(extra="forbid") @@ -499,9 +548,11 @@ TypedExpression.model_rebuild() __all__ = [ "DatasetDefinition", + "DefinitionGovernance", "DefinitionUpdateRequest", "DefinitionWriteRequest", "DimensionDefinition", + "DrillContextCreateRequest", "FilterClause", "ImportAssessmentRequest", "MeasureDefinition", diff --git a/tests/test_manifest.py b/tests/test_manifest.py index 80855a0..7113759 100644 --- a/tests/test_manifest.py +++ b/tests/test_manifest.py @@ -14,7 +14,7 @@ class ReportingManifestTests(unittest.TestCase): self.assertEqual("@govoplan/reporting-webui", manifest.frontend.package_name) self.assertIsNotNone(manifest.route_factory) self.assertIsNotNone(manifest.migration_spec) - self.assertEqual(5, len(manifest.provides_interfaces)) + self.assertEqual(7, len(manifest.provides_interfaces)) self.assertEqual(1, len(manifest.search_sources)) self.assertIn("dataflow", manifest.optional_dependencies) self.assertIn("policy", manifest.optional_dependencies) diff --git a/tests/test_migrations.py b/tests/test_migrations.py index b0994e0..a341d53 100644 --- a/tests/test_migrations.py +++ b/tests/test_migrations.py @@ -26,6 +26,7 @@ def test_fresh_migration_creates_provider_evidence_tables_and_current_head() -> tables = set(inspect(connection).get_table_names()) assert { "reporting_provider_executions", + "reporting_drill_contexts", "reporting_provider_exports", }.issubset(tables) assert "b7c4e1a9d2f6" in set( diff --git a/tests/test_module_permutations.py b/tests/test_module_permutations.py index dea2d82..07ad650 100644 --- a/tests/test_module_permutations.py +++ b/tests/test_module_permutations.py @@ -70,6 +70,34 @@ assert provider.contract_version == '1.0' _run_probe(script) +def test_reporting_starts_without_files_or_mail_and_keeps_targets_optional() -> None: + script = """ +import importlib.abc +import sys + +class Blocker(importlib.abc.MetaPathFinder): + def find_spec(self, fullname, path=None, target=None): + if fullname == 'govoplan_files' or fullname.startswith('govoplan_files.'): + raise ModuleNotFoundError("Files is physically absent", name=fullname) + if fullname == 'govoplan_mail' or fullname.startswith('govoplan_mail.'): + raise ModuleNotFoundError("Mail is physically absent", name=fullname) + return None + +sys.meta_path.insert(0, Blocker()) +from govoplan_reporting.backend.contracts import ( + CAPABILITY_REPORTING_PUBLICATION_FILES, + CAPABILITY_REPORTING_PUBLICATION_MAIL, +) +from govoplan_reporting.backend.manifest import get_manifest +manifest = get_manifest() +assert 'files' in manifest.optional_dependencies +assert 'mail' in manifest.optional_dependencies +assert CAPABILITY_REPORTING_PUBLICATION_FILES in manifest.capability_factories +assert CAPABILITY_REPORTING_PUBLICATION_MAIL in manifest.capability_factories +""" + _run_probe(script) + + def _run_probe(source: str) -> None: environment = dict(os.environ) environment["PYTHONPATH"] = os.pathsep.join( diff --git a/tests/test_reporting_service.py b/tests/test_reporting_service.py index 8386abe..83f8599 100644 --- a/tests/test_reporting_service.py +++ b/tests/test_reporting_service.py @@ -8,6 +8,10 @@ from sqlalchemy import create_engine from sqlalchemy.orm import Session from govoplan_core.db.base import Base +from govoplan_core.core.files import ( + CAPABILITY_FILES_ARTIFACT_STORE, + ManagedArtifactRef, +) from govoplan_core.security.module_permissions import scopes_grant_compatible from govoplan_reporting.backend.definitions import ( ADMIN_SCOPE, @@ -27,6 +31,11 @@ from govoplan_reporting.backend.execution import ( execute_report, run_quality_plan, ) +from govoplan_reporting.backend.drilldown import ( + ReportingDrillError, + create_drill_context, + resolve_drill_context, +) from govoplan_reporting.backend.operations import ( IMPORT_SCOPE, PUBLISH_SCOPE, @@ -35,10 +44,24 @@ from govoplan_reporting.backend.operations import ( assess_import, dispatch_due_schedules, export_execution, + list_publications, + publish_execution, upsert_saved_view, upsert_schedule, ) from govoplan_reporting.backend.schemas import ReportQuery +from govoplan_reporting.backend.postgres_planner import compile_postgres_query +from govoplan_reporting.backend.contracts import ( + CAPABILITY_REPORTING_PUBLICATION_FILES, +) +from govoplan_reporting.backend.publication_targets import ( + FilesReportingPublicationTarget, + publication_target_catalog, +) +from govoplan_reporting.backend.schemas import ( + DatasetDefinition, + SemanticModelDefinition, +) NOW = datetime(2026, 8, 1, 10, 0, tzinfo=UTC) @@ -79,6 +102,36 @@ class Principal: return scopes_grant_compatible(self.scopes, scope) +class CapabilityRegistry: + def __init__(self) -> None: + self.providers: dict[str, object] = {} + + def has_capability(self, name: str) -> bool: + return name in self.providers + + def capability(self, name: str) -> object | None: + return self.providers.get(name) + + +class ArtifactStore: + def __init__(self) -> None: + self.requests: list[object] = [] + + def store_artifact(self, session, principal, *, request): + del session, principal + self.requests.append(request) + return ManagedArtifactRef( + file_asset_id="asset-1", + file_version_id="version-1", + filename=request.filename, + display_path=f"{request.folder}/{request.filename}", + content_type=request.content_type, + size_bytes=len(request.payload), + sha256="a" * 64, + provenance={"stored": True}, + ) + + class ReportingServiceTests(unittest.TestCase): def setUp(self) -> None: self.engine = create_engine("sqlite+pysqlite:///:memory:") @@ -404,6 +457,216 @@ class ReportingServiceTests(unittest.TestCase): accepted_approximations=[], ) + def test_drill_context_is_bounded_actor_bound_and_reauthorized(self) -> None: + self._create_report_graph() + execution = execute_report( + self.session, + self.principal, + registry=None, + report_id="report-1", + report_revision=1, + parameters={}, + query=None, + idempotency_key="drill-source", + ) + north = next(row for row in execution["rows"] if row["region"] == "North") + context = create_drill_context( + self.session, + self.principal, + registry=None, + execution_id=str(execution["execution_id"]), + aggregate_row=north, + limit=50, + ) + detail = resolve_drill_context( + self.session, + self.principal, + registry=None, + token=str(context["token"]), + ) + self.assertEqual(2, detail["total_rows"]) + self.assertEqual({"North"}, {row["region"] for row in detail["rows"]}) + self.assertEqual("region", detail["dimension_path"][0]["dimension"]) + with self.assertRaises(PermissionError): + resolve_drill_context( + self.session, + Principal(account_id="another-analyst"), + registry=None, + token=str(context["token"]), + ) + with self.assertRaises(ReportingDrillError): + create_drill_context( + self.session, + self.principal, + registry=None, + execution_id=str(execution["execution_id"]), + aggregate_row={"region": "Not an execution row"}, + limit=50, + ) + + def test_governance_scope_inheritance_never_broadens_parent_limits(self) -> None: + system = Principal( + scopes=(*ALL_SCOPES, "system:governance:write"), + group_ids=("group-reporters",), + ) + dataset = dataset_payload() + dataset["governance"] = { + "scope_type": "system", + "inherit_to_lower_scopes": True, + "allow_run": True, + "allow_reuse": True, + "allow_automation": False, + } + self._create("dataset", "dataset-governed", dataset, principal=system) + semantic = semantic_payload(dataset_id="dataset-governed") + semantic["governance"] = { + "scope_type": "tenant", + "inherit_to_lower_scopes": True, + "allow_run": True, + "allow_reuse": True, + "allow_automation": True, + } + with self.assertRaisesRegex(ValueError, "cannot broaden inherited limits"): + self._create( + "semantic_model", + "semantic-broadened", + semantic, + principal=system, + ) + semantic["governance"]["allow_automation"] = False + semantic_record = self._create( + "semantic_model", + "semantic-governed", + semantic, + principal=system, + ) + semantic_governance = semantic_record.payload["governance"] + self.assertEqual("system", semantic_governance["source_scope"]["scope_type"]) + self.assertFalse( + semantic_governance["source_effective_limits"]["allow_automation"] + ) + report = report_payload() + report["semantic_model_id"] = "semantic-governed" + report["governance"] = { + "scope_type": "group", + "scope_id": "group-reporters", + "inherit_to_lower_scopes": False, + "allow_run": True, + "allow_reuse": False, + "allow_automation": False, + } + self._create("report", "report-governed", report, principal=system) + self.assertIsNotNone( + get_definition( + self.session, + system, + definition_kind="report", + definition_id="report-governed", + ) + ) + self.assertIsNone( + get_definition( + self.session, + Principal(account_id="outsider"), + definition_kind="report", + definition_id="report-governed", + ) + ) + + def test_postgres_plan_is_bounded_and_parameterized(self) -> None: + dataset = DatasetDefinition.model_validate(dataset_payload()) + semantic = SemanticModelDefinition.model_validate(semantic_payload()) + query = ReportQuery.model_validate( + { + "mode": "summary", + "dimensions": ["region"], + "measures": ["amount", "value_per_case"], + "filters": [ + { + "dimension": "region", + "operator": "contains", + "value": "North%' OR TRUE --", + } + ], + "sort": [{"key": "amount", "direction": "desc"}], + "limit": 25, + } + ) + plan = compile_postgres_query(dataset, semantic, query) + self.assertIn("GROUP BY", plan.sql) + self.assertIn("LIMIT :result_limit OFFSET :result_offset", plan.sql) + self.assertNotIn("North%' OR TRUE --", plan.sql) + self.assertIn("North", str(plan.parameters["filter_0"])) + calculated_only = compile_postgres_query( + dataset, + semantic, + ReportQuery( + mode="summary", + dimensions=["region"], + measures=["value_per_case"], + ), + ) + self.assertIn('SUM(NULLIF(source_row ->> :measure_0, \'\')::numeric)', calculated_only.sql) + self.assertIn('AS "value_per_case"', calculated_only.sql) + + def test_files_publication_is_idempotent_and_retains_evidence(self) -> None: + self._create_report_graph() + registry = CapabilityRegistry() + store = ArtifactStore() + registry.providers[CAPABILITY_FILES_ARTIFACT_STORE] = store + registry.providers[CAPABILITY_REPORTING_PUBLICATION_FILES] = ( + FilesReportingPublicationTarget(registry) + ) + execution = execute_report( + self.session, + self.principal, + registry=registry, + report_id="report-1", + report_revision=1, + parameters={}, + query=None, + idempotency_key="publish-source", + ) + first = publish_execution( + self.session, + self.principal, + registry=registry, + execution_id=str(execution["execution_id"]), + target_capability=CAPABILITY_REPORTING_PUBLICATION_FILES, + target_ref="Reports/Monthly", + format="csv", + idempotency_key="publish-files-once", + options={"filename": "regional workload.csv"}, + ) + replay = publish_execution( + self.session, + self.principal, + registry=registry, + execution_id=str(execution["execution_id"]), + target_capability=CAPABILITY_REPORTING_PUBLICATION_FILES, + target_ref="Reports/Monthly", + format="csv", + idempotency_key="publish-files-once", + options={"filename": "regional workload.csv"}, + ) + self.assertEqual(first["publication_id"], replay["publication_id"]) + self.assertEqual(1, len(store.requests)) + self.assertEqual("version-1", first["evidence"]["file_version_id"]) + self.assertEqual( + 1, + len( + list_publications( + self.session, + self.principal, + execution_id=str(execution["execution_id"]), + ) + ), + ) + targets = publication_target_catalog(registry) + self.assertTrue(targets[0]["available"]) + self.assertFalse(targets[1]["available"]) + self.assertIn("Enable Mail", str(targets[1]["reason"])) + def _create_report_graph( self, *, diff --git a/webui/scripts/test-interface-pattern.mjs b/webui/scripts/test-interface-pattern.mjs index a92f33f..81bb7bc 100644 --- a/webui/scripts/test-interface-pattern.mjs +++ b/webui/scripts/test-interface-pattern.mjs @@ -9,6 +9,9 @@ assert.ok(page.includes("DocumentationHelpLink"), "Reporting exposes configured- assert.ok(page.includes("PageScrollViewport"), "Reporting owns bounded catalogue and inspector scrolling"); assert.ok(page.includes("DataGrid"), "Tabular report results use the shared grid"); assert.ok(page.includes("; fallback_reason?: string | null; }; + delivery_authorization?: Record; +}; + +export type ReportingDrillContext = { + token: string; + drill_context_id: string; + execution_id: string; + dimension_path: Array<{ dimension: string; label: string; value: unknown }>; + expires_at: string; +}; + +export type ReportingDrillResult = Omit & { + rows: Array>; + schema: Array<{ name: string; type: string }>; + total_rows: number; + truncated: boolean; + source_fingerprints: Array>; + policy_provenance: Record; +}; + +export type ReportingSchedule = { + schedule_id: string; + report_id: string; + report_revision: number; + name: string; + revision: number; + trigger_kind: "scheduled" | "interval"; + trigger_config: Record; + parameters: Record; + query: ReportingQuery; + publication_target: Record; + enabled: boolean; + next_run_at?: string | null; + last_run_at?: string | null; + last_execution_id?: string | null; +}; + +export type ReportingPublicationTarget = { + capability: string; + label: string; + available: boolean; + reason?: string | null; + formats: string[]; + target_label: string; + target_required: boolean; + required_options: string[]; +}; + +export type ReportingPublication = { + publication_id: string; + execution_id: string; + target_capability: string; + target_ref?: string | null; + format: string; + status: string; + evidence: Record; + error?: string | null; + completed_at?: string | null; }; export type ReportingSavedView = { @@ -307,6 +367,26 @@ export function listExecutions( return apiFetch(settings, `/api/v1/reporting/reports/${encodeURIComponent(reportId)}/executions?limit=30`, { signal }); } +export function createDrillContext( + settings: ApiSettings, + executionId: string, + aggregateRow: Record, + limit = 200 +): Promise { + return apiFetch(settings, `/api/v1/reporting/executions/${encodeURIComponent(executionId)}/drill-contexts`, { + method: "POST", + body: JSON.stringify({ aggregate_row: aggregateRow, limit }) + }); +} + +export function resolveDrillContext( + settings: ApiSettings, + token: string, + signal?: AbortSignal +): Promise { + return apiFetch(settings, `/api/v1/reporting/drill-contexts/${encodeURIComponent(token)}`, { signal }); +} + export function listSavedViews( settings: ApiSettings, reportId: string, @@ -364,6 +444,72 @@ export function createIntervalSchedule( }); } +export function listSchedules( + settings: ApiSettings, + reportId: string, + signal?: AbortSignal +): Promise<{ schedules: ReportingSchedule[] }> { + return apiFetch(settings, apiPath("/api/v1/reporting/schedules", { report_id: reportId }), { signal }); +} + +export function updateSchedule( + settings: ApiSettings, + schedule: ReportingSchedule, + changes: Partial> +): Promise { + return apiFetch(settings, `/api/v1/reporting/schedules/${encodeURIComponent(schedule.schedule_id)}`, { + method: "PUT", + body: JSON.stringify({ + schedule_id: schedule.schedule_id, + report_id: schedule.report_id, + report_revision: schedule.report_revision, + name: changes.name ?? schedule.name, + trigger_kind: schedule.trigger_kind, + trigger_config: schedule.trigger_config, + parameters: schedule.parameters, + query: schedule.query, + publication_target: schedule.publication_target, + enabled: changes.enabled ?? schedule.enabled, + next_run_at: schedule.next_run_at ?? null, + expected_revision: schedule.revision + }) + }); +} + +export function listPublicationTargets( + settings: ApiSettings, + signal?: AbortSignal +): Promise<{ targets: ReportingPublicationTarget[] }> { + return apiFetch(settings, "/api/v1/reporting/publication-targets", { signal }); +} + +export function listPublications( + settings: ApiSettings, + executionId: string, + signal?: AbortSignal +): Promise<{ publications: ReportingPublication[] }> { + return apiFetch(settings, apiPath("/api/v1/reporting/publications", { execution_id: executionId }), { signal }); +} + +export function publishExecution( + settings: ApiSettings, + executionId: string, + request: { + target_capability: string; + target_ref?: string | null; + format: string; + options: Record; + } +): Promise { + return apiFetch(settings, `/api/v1/reporting/executions/${encodeURIComponent(executionId)}/publications`, { + method: "POST", + body: JSON.stringify({ + ...request, + idempotency_key: crypto.randomUUID() + }) + }); +} + export async function downloadExecution( settings: ApiSettings, executionId: string, diff --git a/webui/src/features/reporting/ReportingPage.tsx b/webui/src/features/reporting/ReportingPage.tsx index 4acb6aa..f25b8fb 100644 --- a/webui/src/features/reporting/ReportingPage.tsx +++ b/webui/src/features/reporting/ReportingPage.tsx @@ -1,8 +1,10 @@ import { BarChart3, CalendarClock, + ChevronRight, Download, FileJson, + FolderOutput, History, Play, RefreshCw, @@ -28,27 +30,39 @@ import { PageScrollViewport, SegmentedControl, StatusBadge, + ToggleSwitch, hasScope, type DataGridColumn, type PlatformRouteContext } from "@govoplan/core-webui"; import { createIntervalSchedule, + createDrillContext, downloadExecution, getDefinition, listDefinitions, listExecutions, + listPublicationTargets, + listPublications, listProviderReports, listSavedViews, + listSchedules, + publishExecution, reportPayload, runReport, saveView, semanticPayload, + resolveDrillContext, + updateSchedule, type ReportExecution, + type ReportingDrillResult, type ReportingDefinition, type ReportingQuery, type ReportingQueryMode, + type ReportingPublication, + type ReportingPublicationTarget, type ReportingSavedView, + type ReportingSchedule, type ProviderReportDescriptor, type SemanticModelPayload } from "../../api/reporting"; @@ -73,14 +87,22 @@ export default function ReportingPage({ settings, auth }: PlatformRouteContext) const [execution, setExecution] = useState(null); const [history, setHistory] = useState([]); const [savedViews, setSavedViews] = useState([]); + const [schedules, setSchedules] = useState([]); + const [publicationTargets, setPublicationTargets] = useState([]); + const [publications, setPublications] = useState([]); const [outputMode, setOutputMode] = useState("visual"); const [loading, setLoading] = useState(true); const [running, setRunning] = useState(false); const [error, setError] = useState(""); const [saveDialogOpen, setSaveDialogOpen] = useState(false); const [scheduleDialogOpen, setScheduleDialogOpen] = useState(false); + const [publishDialogOpen, setPublishDialogOpen] = useState(false); + const [drillDialogOpen, setDrillDialogOpen] = useState(false); + const [drillResult, setDrillResult] = useState(null); + const [drilling, setDrilling] = useState(false); const canRun = hasScope(auth, "reporting:report:run"); const canSchedule = hasScope(auth, "reporting:schedule:write"); + const canPublish = hasScope(auth, "reporting:report:publish"); const selected = useMemo( () => reports.find((item) => item.definition_id === selectedId) ?? null, @@ -102,15 +124,17 @@ export default function ReportingPage({ settings, auth }: PlatformRouteContext) query: submittedSearch, limit: 200 }, signal), - listProviderReports(settings, signal) + listProviderReports(settings, signal), + canPublish ? listPublicationTargets(settings, signal) : Promise.resolve({ targets: [] }) ]). - then(([result, providerResult]) => { + then(([result, providerResult, targetResult]) => { const providerRows = providerResult.reports.filter((item) => { const query = submittedSearch.toLocaleLowerCase(); return !query || `${item.title} ${item.summary} ${item.provider_id}`.toLocaleLowerCase().includes(query); }); setReports(result.definitions); setProviderReports(providerRows); + setPublicationTargets(targetResult.targets); const currentSemanticAvailable = result.definitions.some((item) => item.definition_id === selectedId); const currentProviderAvailable = providerRows.some((item) => `${item.provider_id}:${item.report_id}` === selectedProviderKey); if (!currentSemanticAvailable && !currentProviderAvailable) { @@ -141,6 +165,7 @@ export default function ReportingPage({ settings, auth }: PlatformRouteContext) setExecution(null); setHistory([]); setSavedViews([]); + setSchedules([]); return; } const controller = new AbortController(); @@ -150,12 +175,14 @@ export default function ReportingPage({ settings, auth }: PlatformRouteContext) Promise.all([ getDefinition(settings, "semantic_model", report.semantic_model_id, report.semantic_model_revision, controller.signal), canRun ? listExecutions(settings, selected.definition_id, controller.signal) : Promise.resolve({ executions: [] }), - listSavedViews(settings, selected.definition_id, controller.signal) + listSavedViews(settings, selected.definition_id, controller.signal), + canSchedule ? listSchedules(settings, selected.definition_id, controller.signal) : Promise.resolve({ schedules: [] }) ]). - then(([semanticDefinition, executions, views]) => { + then(([semanticDefinition, executions, views, scheduleResult]) => { setSemantic(semanticPayload(semanticDefinition)); setHistory(executions.executions); setSavedViews(views.views); + setSchedules(scheduleResult.schedules); setExecution(executions.executions.find((item) => item.status === "succeeded") ?? null); }). catch((reason) => { @@ -164,6 +191,20 @@ export default function ReportingPage({ settings, auth }: PlatformRouteContext) return () => controller.abort(); }, [settings, selectedId]); + useEffect(() => { + if (!execution || !canPublish) { + setPublications([]); + return; + } + const controller = new AbortController(); + void listPublications(settings, execution.execution_id, controller.signal). + then((result) => setPublications(result.publications)). + catch((reason) => { + if ((reason as Error).name !== "AbortError") setError(message(reason)); + }); + return () => controller.abort(); + }, [settings, execution?.execution_id, canPublish]); + function submitSearch(event: FormEvent) { event.preventDefault(); setSubmittedSearch(search.trim()); @@ -189,6 +230,23 @@ export default function ReportingPage({ settings, auth }: PlatformRouteContext) if (view.state.query) setQuery(normalizeQuery(view.state.query)); } + async function drill(row: Record) { + if (!execution) return; + setDrillDialogOpen(true); + setDrillResult(null); + setDrilling(true); + setError(""); + try { + const context = await createDrillContext(settings, execution.execution_id, row); + setDrillResult(await resolveDrillContext(settings, context.token)); + } catch (reason) { + setError(message(reason)); + setDrillDialogOpen(false); + } finally { + setDrilling(false); + } + } + return (
@@ -302,13 +360,16 @@ export default function ReportingPage({ settings, auth }: PlatformRouteContext) {execution.total_rows} rows{execution.truncated ? " (truncated)" : ""} } variant="ghost" onClick={() => void downloadExecution(settings, execution.execution_id, "csv").catch((reason) => setError(message(reason)))} /> } variant="ghost" onClick={() => void downloadExecution(settings, execution.execution_id, "json").catch((reason) => setError(message(reason)))} /> + {canPublish && + } variant="ghost" onClick={() => setPublishDialogOpen(true)} /> + } }
{!execution &&
Run the report or select a previous execution.
} - {execution && outputMode === "visual" && } - {execution && outputMode === "table" && } + {execution && outputMode === "visual" && } + {execution && outputMode === "table" && }
:
Select a report.
@@ -322,8 +383,18 @@ export default function ReportingPage({ settings, auth }: PlatformRouteContext) execution={execution} history={history} savedViews={savedViews} + schedules={schedules} + publications={publications} onSelectExecution={setExecution} onApplySavedView={applySavedView} + onScheduleEnabledChange={async (schedule, enabled) => { + try { + const updated = await updateSchedule(settings, schedule, { enabled }); + setSchedules((current) => current.map((item) => item.schedule_id === updated.schedule_id ? updated : item)); + } catch (reason) { + setError(message(reason)); + } + }} />} @@ -343,10 +414,28 @@ export default function ReportingPage({ settings, auth }: PlatformRouteContext) onClose={() => setScheduleDialogOpen(false)} onSave={async (name, seconds) => { if (!selected) return; - await createIntervalSchedule(settings, selected, name, seconds, query, parameters); + const created = await createIntervalSchedule(settings, selected, name, seconds, query, parameters) as ReportingSchedule; + setSchedules((current) => [...current, created].sort((left, right) => left.name.localeCompare(right.name))); setScheduleDialogOpen(false); }} /> + setPublishDialogOpen(false)} + onPublish={async (request) => { + if (!execution) return; + const publication = await publishExecution(settings, execution.execution_id, request); + setPublications((current) => [publication, ...current]); + setPublishDialogOpen(false); + }} + /> + setDrillDialogOpen(false)} + />
); } @@ -446,11 +535,11 @@ function QueryControls({ query, semantic, parameters, parameterValues, onQueryCh ); } -function ReportTable({ execution }: { execution: ReportExecution }) { +function ReportTable({ execution, onDrill }: { execution: ReportExecution; onDrill?: (row: Record) => void }) { const [page, setPage] = useState(0); useEffect(() => setPage(0), [execution.execution_id]); - const columns = useMemo>[]>(() => - execution.schema.map((field) => ({ + const columns = useMemo>[]>(() => { + const result = execution.schema.map((field) => ({ id: field.name, header: humanize(field.name), width: "1fr", @@ -461,7 +550,26 @@ function ReportTable({ execution }: { execution: ReportExecution }) { filterType: field.type === "integer" || field.type === "number" ? field.type : "text", value: (row) => row[field.name], render: (row) => formatValue(row[field.name]) - })), [execution]); + } satisfies DataGridColumn>)); + if (onDrill) { + result.push({ + id: "drill", + header: "Detail", + width: 74, + minWidth: 74, + maxWidth: 74, + render: (row) => ( + } + variant="ghost" + onClick={() => onDrill(row)} + /> + ) + }); + } + return result; + }, [execution, onDrill]); return ( ) => void }) { const visual = execution.visualization; - if (!visual || visual.kind === "table" || !visual.category || !visual.measures?.length) { + const needsCategory = visual?.kind !== "metric"; + if (!visual || visual.kind === "table" || !visual.measures?.length || (needsCategory && !visual.category)) { return (
{visual?.fallback_reason && {visual.fallback_reason}} - +
); } const measure = visual.measures[0]; const values = execution.rows.map((row) => Number(row[measure] ?? 0)); const maximum = Math.max(...values.map((value) => Math.abs(value)), 1); - return ( -
- {execution.rows.map((row, index) => -
- {formatValue(row[visual.category ?? ""])} -
- {formatValue(row[measure])} + if (visual.kind === "metric") { + return ( +
+ {visual.measures.map((key) => +
+ {humanize(key)} + {formatValue(execution.rows[0]?.[key])} +
+ )} +
+
+ ); + } + if (visual.kind === "column") { + return ( +
+
+ {execution.rows.slice(0, 50).map((row, index) => +
+ {formatValue(row[measure])} + + {formatValue(row[visual.category ?? ""])} +
+ )}
- )} -
+
+
+ ); + } + if (visual.kind === "line" || visual.kind === "area") { + const points = chartPoints(values.slice(0, 50), 700, 250); + return ( +
+ + {visual.kind === "area" && } + + +
+ {execution.rows.slice(0, 50).map((row, index) => {formatValue(row[visual.category ?? ""])})} +
+
+
+ ); + } + if (visual.kind === "pie" || visual.kind === "donut") { + const positive = values.map((value) => Math.max(0, value)); + const total = positive.reduce((sum, value) => sum + value, 0) || 1; + const stops = pieStops(positive, total); + return ( +
+
+
    + {execution.rows.slice(0, 12).map((row, index) =>
  1. {formatValue(row[visual.category ?? ""])}{formatValue(row[measure])}
    2. )} +
+
+
+ ); + } + return ( +
+
+ {execution.rows.map((row, index) => +
+ {formatValue(row[visual.category ?? ""])} +
+ {formatValue(row[measure])} +
+ )} +
+
); } -function Inspector({ selected, execution, history, savedViews, onSelectExecution, onApplySavedView }: { +function AccessExplanation({ execution }: { execution: ReportExecution }) { + const provenance = objectValue(execution.provenance); + const explanation = objectValue(provenance.access_explanation); + const hiddenDimensions = stringValues(explanation.hidden_dimensions); + const hiddenMeasures = stringValues(explanation.hidden_measures); + const disabledActions = stringValues(explanation.disabled_actions); + const hiddenRows = Number(explanation.hidden_rows ?? 0); + const reasons = objectValue(explanation.reasons); + if (!hiddenDimensions.length && !hiddenMeasures.length && !disabledActions.length && hiddenRows <= 0) { + return No report fields, rows, or actions were hidden by effective policy.; + } + return ( +
+ Effective access + {hiddenDimensions.length > 0 && Hidden dimensions: {hiddenDimensions.join(", ")}} + {hiddenMeasures.length > 0 && Hidden measures: {hiddenMeasures.join(", ")}} + {hiddenRows > 0 && {hiddenRows} source rows were removed before planning.} + {disabledActions.map((action) => {String(reasons[action] ?? `The ${action} action is disabled by policy.`)})} +
+ ); +} + +function PublishDialog({ open, targets, onClose, onPublish }: { + open: boolean; + targets: ReportingPublicationTarget[]; + onClose: () => void; + onPublish: (request: { target_capability: string; target_ref?: string | null; format: string; options: Record }) => Promise; +}) { + const firstAvailable = targets.find((item) => item.available) ?? targets[0]; + const [targetCapability, setTargetCapability] = useState(firstAvailable?.capability ?? ""); + const [targetRef, setTargetRef] = useState(""); + const [format, setFormat] = useState(firstAvailable?.formats[0] ?? "csv"); + const [filename, setFilename] = useState(""); + const [mailProfileId, setMailProfileId] = useState(""); + const [fromAddress, setFromAddress] = useState(""); + const [subject, setSubject] = useState(""); + const [saving, setSaving] = useState(false); + const [dialogError, setDialogError] = useState(""); + const target = targets.find((item) => item.capability === targetCapability) ?? firstAvailable; + + useEffect(() => { + if (!open) return; + const next = targets.find((item) => item.available) ?? targets[0]; + setTargetCapability(next?.capability ?? ""); + setFormat(next?.formats[0] ?? "csv"); + setTargetRef(""); + setFilename(""); + setMailProfileId(""); + setFromAddress(""); + setSubject(""); + setDialogError(""); + }, [open, targets]); + + const mailTarget = target?.capability.endsWith(".mail") === true; + const valid = Boolean(target?.available) && (!target?.target_required || targetRef.trim()) && (!mailTarget || (mailProfileId.trim() && fromAddress.trim())); + return ( + + + + }> + {dialogError && {dialogError}} +
+ + + {target && } + {!mailTarget && } + {mailTarget && <> + + + + } +
+ {target?.reason && {target.reason}} + + ); +} + +function DrillDialog({ open, loading, result, onClose }: { + open: boolean; + loading: boolean; + result: ReportingDrillResult | null; + onClose: () => void; +}) { + const [page, setPage] = useState(0); + useEffect(() => setPage(0), [result?.drill_context_id]); + const columns = useMemo>[]>(() => + (result?.schema ?? []).map((field) => ({ + id: field.name, + header: humanize(field.name), + width: "1fr", + minWidth: 120, + resizable: true, + sortable: true, + filterable: true, + filterType: field.type === "number" || field.type === "integer" ? field.type : "text", + value: (row) => row[field.name], + render: (row) => formatValue(row[field.name]) + })), [result]); + return ( + Close}> + {loading && } + {result && <> + +
+ `${result.drill_context_id}:${index}`} + initialFit="container" + resizeBehavior="cover" + emptyText="No contributing rows are authorized." + pagination={{ page, pageSize: 50, onPageChange: setPage }} + /> +
+ {result.total_rows} authorized rows{result.truncated ? " (bounded result)" : ""}. Access and source fingerprints were rechecked for this drill. + } + + ); +} + +function Inspector({ selected, execution, history, savedViews, schedules, publications, onSelectExecution, onApplySavedView, onScheduleEnabledChange }: { selected: ReportingDefinition | null; execution: ReportExecution | null; history: ReportExecution[]; savedViews: ReportingSavedView[]; + schedules: ReportingSchedule[]; + publications: ReportingPublication[]; onSelectExecution: (execution: ReportExecution) => void; onApplySavedView: (view: ReportingSavedView) => void; + onScheduleEnabledChange: (schedule: ReportingSchedule, enabled: boolean) => void; }) { return (
@@ -524,6 +860,29 @@ function Inspector({ selected, execution, history, savedViews, onSelectExecution )} + {schedules.length > 0 && +
+

Schedules

+ {schedules.map((schedule) => +
+ {schedule.name}{schedule.trigger_kind === "interval" ? `Every ${formatInterval(schedule.trigger_config.seconds)}` : "Scheduled"} + onScheduleEnabledChange(schedule, enabled)} /> +
+ )} +
+ } + {publications.length > 0 && +
+

Publications

+ {publications.map((publication) => +
+ {humanize(publication.target_capability.split(".").at(-1) ?? "target")} + + {publication.completed_at ? formatDateTime(publication.completed_at) : "Pending"} +
+ )} +
+ }

Saved views

{savedViews.length === 0 &&

No saved views.

} @@ -548,6 +907,7 @@ function Inspector({ selected, execution, history, savedViews, onSelectExecution {item.message ?? item.code ?? "Execution diagnostic"} )} + {execution && }
}
@@ -619,6 +979,48 @@ function formatDateTime(value: string): string { return Number.isNaN(parsed.valueOf()) ? value : new Intl.DateTimeFormat(undefined, { dateStyle: "medium", timeStyle: "short" }).format(parsed); } +function formatInterval(value: unknown): string { + const seconds = Number(value); + if (seconds === 3600) return "hour"; + if (seconds === 86400) return "day"; + if (seconds === 604800) return "week"; + if (seconds === 2592000) return "30 days"; + return `${Number.isFinite(seconds) ? seconds : 0} seconds`; +} + +function chartPoints(values: number[], width: number, height: number): string { + if (!values.length) return ""; + const finite = values.map((value) => Number.isFinite(value) ? value : 0); + const minimum = Math.min(...finite); + const maximum = Math.max(...finite); + const range = maximum - minimum || 1; + const divisor = Math.max(1, finite.length - 1); + return finite.map((value, index) => { + const x = index / divisor * width; + const y = height - ((value - minimum) / range * (height - 20) + 10); + return `${x.toFixed(2)},${y.toFixed(2)}`; + }).join(" "); +} + +const PIE_COLORS = ["#2f7d6e", "#3366a8", "#c28b2c", "#9a4f71", "#5f7f3a", "#b85c3b", "#586176", "#2e8b9a"]; + +function pieStops(values: number[], total: number): string { + let offset = 0; + return values.slice(0, 12).map((value, index) => { + const start = offset; + offset += value / total * 100; + return `${PIE_COLORS[index % PIE_COLORS.length]} ${start.toFixed(2)}% ${offset.toFixed(2)}%`; + }).join(", "); +} + +function objectValue(value: unknown): Record { + return value && typeof value === "object" && !Array.isArray(value) ? value as Record : {}; +} + +function stringValues(value: unknown): string[] { + return Array.isArray(value) ? value.map(String) : []; +} + function humanize(value: string): string { return value.replaceAll("_", " ").replace(/\b\w/g, (letter) => letter.toUpperCase()); } diff --git a/webui/src/features/reporting/ReportingReportsWidget.tsx b/webui/src/features/reporting/ReportingReportsWidget.tsx new file mode 100644 index 0000000..45438bf --- /dev/null +++ b/webui/src/features/reporting/ReportingReportsWidget.tsx @@ -0,0 +1,57 @@ +import { useCallback } from "react"; +import { BarChart3 } from "lucide-react"; +import { Link } from "react-router"; +import { + DashboardWidgetList, + DismissibleAlert, + LoadingFrame, + StatusBadge, + useDashboardWidgetData, + type ApiSettings, + type DashboardWidgetConfiguration +} from "@govoplan/core-webui"; +import { listDefinitions } from "../../api/reporting"; + + +export default function ReportingReportsWidget({ settings, refreshKey, configuration }: { + settings: ApiSettings; + refreshKey: number; + configuration: DashboardWidgetConfiguration; +}) { + const maxItems = boundedNumber(configuration.maxItems, 5, 1, 12); + const load = useCallback(async () => { + const result = await listDefinitions(settings, { + kinds: ["report"], + status: ["active"], + limit: maxItems + }); + return result.definitions.slice(0, maxItems); + }, [maxItems, settings]); + const { data, loading, error } = useDashboardWidgetData(load, refreshKey); + return ( + + {error && {error}} + ({ + id: report.definition_id, + title: report.name, + detail: report.description || report.definition_key, + meta: `Revision ${report.revision}`, + leading: + ); +} + + +function boundedNumber(value: unknown, fallback: number, minimum: number, maximum: number): number { + const numeric = typeof value === "number" ? value : Number(value); + return Number.isFinite(numeric) ? Math.max(minimum, Math.min(maximum, Math.round(numeric))) : fallback; +} diff --git a/webui/src/module.ts b/webui/src/module.ts index efdbdd5..73fb07d 100644 --- a/webui/src/module.ts +++ b/webui/src/module.ts @@ -1,10 +1,46 @@ import { createElement, lazy } from "react"; -import type { PlatformWebModule } from "@govoplan/core-webui"; +import type { DashboardWidgetsUiCapability, PlatformWebModule } from "@govoplan/core-webui"; +import ReportingReportsWidget from "./features/reporting/ReportingReportsWidget"; import "./styles/reporting.css"; const ReportingPage = lazy(() => import("./features/reporting/ReportingPage")); +const reportingDashboardWidgets: DashboardWidgetsUiCapability = { + widgets: [ + { + id: "reporting.reports", + surfaceId: "reporting.widget.reports", + title: "Reports", + description: "Active governed reports available in the current scope.", + moduleId: "reporting", + category: "Analysis", + order: 75, + defaultVisible: false, + defaultSize: "medium", + supportedSizes: ["medium", "wide"], + anyOf: ["reporting:definition:read"], + refreshIntervalMs: 60_000, + defaultConfiguration: { maxItems: 5 }, + configurationFields: [ + { + id: "maxItems", + label: "Maximum reports", + kind: "number", + min: 1, + max: 12, + step: 1, + required: true + } + ], + render: ({ settings, refreshKey, configuration }) => createElement( + ReportingReportsWidget, + { settings, refreshKey, configuration } + ) + } + ] +}; + export const reportingModule: PlatformWebModule = { id: "reporting", label: "Reporting", @@ -52,8 +88,12 @@ export const reportingModule: PlatformWebModule = { { id: "reporting.navigation", moduleId: "reporting", kind: "navigation", label: "Reporting navigation", order: 10 }, { id: "reporting.workspace", moduleId: "reporting", kind: "route", label: "Reporting workspace", order: 20 }, { id: "reporting.parameters", moduleId: "reporting", kind: "section", label: "Report parameters and filters", parentId: "reporting.workspace", order: 30 }, - { id: "reporting.results", moduleId: "reporting", kind: "section", label: "Authorized report results", parentId: "reporting.workspace", order: 40 } - ] + { id: "reporting.results", moduleId: "reporting", kind: "section", label: "Authorized report results", parentId: "reporting.workspace", order: 40 }, + { id: "reporting.widget.reports", moduleId: "reporting", kind: "section", label: "Reports dashboard widget", order: 75 } + ], + uiCapabilities: { + "dashboard.widgets": reportingDashboardWidgets + } }; export default reportingModule; diff --git a/webui/src/styles/reporting.css b/webui/src/styles/reporting.css index 95588d0..1b716c7 100644 --- a/webui/src/styles/reporting.css +++ b/webui/src/styles/reporting.css @@ -361,6 +361,152 @@ padding: 10px; } +.reporting-chart-stack, +.reporting-line-chart, +.reporting-metric-grid, +.reporting-pie-layout { + display: grid; + gap: 14px; + min-width: 0; +} + +.reporting-column-chart { + display: flex; + align-items: end; + gap: 8px; + min-height: 280px; + padding: 18px 12px 0; + overflow-x: auto; + border-bottom: 1px solid var(--border); +} + +.reporting-column { + display: grid; + grid-template-rows: 24px minmax(180px, 1fr) 32px; + align-items: end; + min-width: 54px; + flex: 1 0 54px; + gap: 4px; + text-align: center; +} + +.reporting-column strong, +.reporting-column span { + overflow: hidden; + font-size: 0.72rem; + text-overflow: ellipsis; + white-space: nowrap; +} + +.reporting-column i { + display: block; + width: 72%; + margin: 0 auto; + border-radius: 3px 3px 0 0; + background: var(--accent); +} + +.reporting-line-chart svg { + width: 100%; + height: 280px; + overflow: visible; + border: 1px solid var(--border); + border-radius: 4px; + background: var(--surface-raised); +} + +.reporting-chart-line { + fill: none; + stroke: var(--accent); + stroke-width: 3; + vector-effect: non-scaling-stroke; +} + +.reporting-chart-area { + fill: color-mix(in srgb, var(--accent) 28%, transparent); +} + +.reporting-chart-labels { + display: flex; + justify-content: space-between; + gap: 8px; + overflow-x: auto; + color: var(--text-soft); + font-size: 0.7rem; +} + +.reporting-metric-grid { + grid-template-columns: repeat(auto-fit, minmax(180px, 1fr)); +} + +.reporting-metric { + display: flex; + min-height: 92px; + flex-direction: column; + justify-content: center; + gap: 6px; + padding: 14px; + border: 1px solid var(--border); + border-radius: 6px; + background: var(--surface-raised); +} + +.reporting-metric span { + color: var(--text-soft); + font-size: 0.76rem; +} + +.reporting-metric strong { + font-size: 1.45rem; +} + +.reporting-pie-layout { + grid-template-columns: minmax(180px, 300px) minmax(240px, 1fr); + align-items: center; +} + +.reporting-pie { + width: min(100%, 280px); + aspect-ratio: 1; + margin: 0 auto; + border-radius: 50%; + box-shadow: inset 0 0 0 1px var(--border); +} + +.reporting-pie.is-donut { + border: 58px solid var(--surface-raised); +} + +.reporting-pie-layout ol { + display: grid; + gap: 7px; + margin: 0; + padding: 0; + list-style: none; +} + +.reporting-pie-layout li { + display: grid; + grid-template-columns: 12px minmax(0, 1fr) auto; + align-items: center; + gap: 8px; +} + +.reporting-swatch { + width: 10px; + height: 10px; + border-radius: 2px; + background: #2f7d6e; +} + +.reporting-swatch-1 { background: #3366a8; } +.reporting-swatch-2 { background: #c28b2c; } +.reporting-swatch-3 { background: #9a4f71; } +.reporting-swatch-4 { background: #5f7f3a; } +.reporting-swatch-5 { background: #b85c3b; } +.reporting-swatch-6 { background: #586176; } +.reporting-swatch-7 { background: #2e8b9a; } + .reporting-bar-row { display: grid; grid-template-columns: minmax(100px, 22%) minmax(180px, 1fr) minmax(80px, auto); @@ -384,11 +530,87 @@ } .reporting-chart-table { + grid-column: 1 / -1; margin-top: 14px; padding-top: 12px; border-top: 1px solid var(--border); } +.reporting-inspector-toggle, +.reporting-inspector-record { + display: grid; + grid-template-columns: minmax(0, 1fr) auto; + align-items: center; + gap: 6px 10px; + padding: 9px 10px; + border-bottom: 1px solid var(--border); +} + +.reporting-inspector-toggle > span, +.reporting-inspector-record > span { + display: flex; + min-width: 0; + flex-direction: column; + gap: 2px; +} + +.reporting-inspector-toggle small, +.reporting-inspector-record small { + grid-column: 1 / -1; +} + +.reporting-inspector-toggle .toggle-switch-copy { + display: none; +} + +.reporting-access-explanation { + display: flex; + flex-direction: column; + gap: 4px; + margin: 8px; + padding: 9px 10px; + border-left: 3px solid var(--accent); + background: var(--hover-bg); + font-size: 0.76rem; +} + +.reporting-dialog-span { + grid-column: 1 / -1; +} + +.reporting-drill-dialog { + width: min(1100px, calc(100vw - 32px)); + height: min(760px, calc(100vh - 32px)); +} + +.reporting-drill-dialog .dialog-body { + display: flex; + min-height: 0; + flex-direction: column; + gap: 10px; + overflow: hidden; +} + +.reporting-drill-path { + display: flex; + flex-wrap: wrap; + gap: 7px; +} + +.reporting-drill-path span { + padding: 5px 8px; + border: 1px solid var(--border); + border-radius: 4px; + background: var(--surface-subtle, var(--surface)); + font-size: 0.76rem; +} + +.reporting-drill-grid { + min-height: 0; + flex: 1; + overflow: auto; +} + .reporting-dialog-grid { display: grid; grid-template-columns: repeat(2, minmax(0, 1fr));