905 lines
29 KiB
Python
905 lines
29 KiB
Python
from __future__ import annotations
|
|
|
|
from collections.abc import Mapping
|
|
import csv
|
|
from datetime import UTC, datetime, timedelta
|
|
from io import StringIO
|
|
import hashlib
|
|
import json
|
|
from typing import Any
|
|
import uuid
|
|
|
|
from sqlalchemy import or_
|
|
from sqlalchemy.orm import Session
|
|
|
|
from govoplan_core.core.events import (
|
|
EventActorRef,
|
|
EventObjectRef,
|
|
EventTenantRef,
|
|
PlatformEvent,
|
|
emit_platform_event,
|
|
)
|
|
from govoplan_core.security.module_permissions import scopes_grant_compatible
|
|
from govoplan_core.security.time import utc_now
|
|
from govoplan_reporting.backend.contracts import (
|
|
ReportingPublicationPayload,
|
|
ReportingPublicationTarget,
|
|
capability,
|
|
)
|
|
from govoplan_reporting.backend.db.models import (
|
|
ReportingImportAssessment,
|
|
ReportingPublication,
|
|
ReportingSavedView,
|
|
ReportingSchedule,
|
|
)
|
|
from govoplan_reporting.backend.definitions import ADMIN_SCOPE, get_definition
|
|
from govoplan_reporting.backend.execution import (
|
|
ReportingExecutionFailure,
|
|
execute_report,
|
|
get_execution,
|
|
)
|
|
from govoplan_reporting.backend.schemas import ReportQuery
|
|
|
|
|
|
PUBLISH_SCOPE = "reporting:report:publish"
|
|
SCHEDULE_SCOPE = "reporting:schedule:write"
|
|
IMPORT_SCOPE = "reporting:import:assess"
|
|
|
|
|
|
class ReportingOperationError(ValueError):
|
|
pass
|
|
|
|
|
|
class SqlReportingScheduler:
|
|
def __init__(self, registry: object | None) -> None:
|
|
self.registry = registry
|
|
|
|
def dispatch_due(
|
|
self,
|
|
session: object,
|
|
principal: object,
|
|
*,
|
|
now: datetime | None = None,
|
|
limit: int = 20,
|
|
) -> Mapping[str, object]:
|
|
return dispatch_due_schedules(
|
|
_session(session),
|
|
principal,
|
|
registry=self.registry,
|
|
now=now,
|
|
limit=limit,
|
|
)
|
|
|
|
|
|
def upsert_saved_view(
|
|
session: Session,
|
|
principal: object,
|
|
*,
|
|
view_id: str,
|
|
report_id: str,
|
|
report_revision: int,
|
|
name: str,
|
|
state: Mapping[str, object],
|
|
shared: bool,
|
|
access: Mapping[str, object],
|
|
expected_revision: int | None,
|
|
) -> dict[str, object]:
|
|
report = get_definition(
|
|
session,
|
|
principal,
|
|
definition_kind="report",
|
|
definition_id=report_id,
|
|
revision=report_revision,
|
|
)
|
|
if report is None:
|
|
raise LookupError("Reporting report definition not found.")
|
|
_validate_saved_view_state(state)
|
|
owner_id = _actor(principal)
|
|
if owner_id is None:
|
|
raise ReportingOperationError("Saved views require an account owner.")
|
|
row = (
|
|
session.query(ReportingSavedView)
|
|
.filter(
|
|
ReportingSavedView.tenant_id == _tenant(principal),
|
|
ReportingSavedView.view_id == view_id,
|
|
)
|
|
.with_for_update()
|
|
.one_or_none()
|
|
)
|
|
if row is None:
|
|
if expected_revision is not None:
|
|
raise ReportingOperationError(
|
|
"Saved-view revision conflict: no view exists."
|
|
)
|
|
row = ReportingSavedView(
|
|
tenant_id=_tenant(principal),
|
|
view_id=_required(view_id, "Saved-view identifier", 36),
|
|
report_id=report_id,
|
|
report_revision=report_revision,
|
|
owner_kind="account",
|
|
owner_id=owner_id,
|
|
name=_required(name, "Saved-view name", 500),
|
|
revision=1,
|
|
state=_json_value(state),
|
|
shared=shared,
|
|
access=_json_value(access),
|
|
)
|
|
session.add(row)
|
|
else:
|
|
if row.owner_id != owner_id and not _has_scope(principal, ADMIN_SCOPE):
|
|
raise PermissionError("Only the saved-view owner or an admin may edit it.")
|
|
if expected_revision != row.revision:
|
|
raise ReportingOperationError(
|
|
"Saved-view revision conflict: the expected revision is stale."
|
|
)
|
|
row.report_id = report_id
|
|
row.report_revision = report_revision
|
|
row.name = _required(name, "Saved-view name", 500)
|
|
row.state = _json_value(state)
|
|
row.shared = shared
|
|
row.access = _json_value(access)
|
|
row.revision += 1
|
|
session.flush()
|
|
return _saved_view_payload(row)
|
|
|
|
|
|
def list_saved_views(
|
|
session: Session,
|
|
principal: object,
|
|
*,
|
|
report_id: str,
|
|
) -> tuple[dict[str, object], ...]:
|
|
if (
|
|
get_definition(
|
|
session,
|
|
principal,
|
|
definition_kind="report",
|
|
definition_id=report_id,
|
|
)
|
|
is None
|
|
):
|
|
return ()
|
|
owner_id = _actor(principal)
|
|
rows = (
|
|
session.query(ReportingSavedView)
|
|
.filter(
|
|
ReportingSavedView.tenant_id == _tenant(principal),
|
|
ReportingSavedView.report_id == report_id,
|
|
or_(
|
|
ReportingSavedView.shared.is_(True),
|
|
ReportingSavedView.owner_id == owner_id,
|
|
),
|
|
)
|
|
.order_by(ReportingSavedView.name.asc())
|
|
.all()
|
|
)
|
|
return tuple(_saved_view_payload(row) for row in rows)
|
|
|
|
|
|
def delete_saved_view(
|
|
session: Session,
|
|
principal: object,
|
|
*,
|
|
view_id: str,
|
|
) -> bool:
|
|
row = (
|
|
session.query(ReportingSavedView)
|
|
.filter(
|
|
ReportingSavedView.tenant_id == _tenant(principal),
|
|
ReportingSavedView.view_id == view_id,
|
|
)
|
|
.with_for_update()
|
|
.one_or_none()
|
|
)
|
|
if row is None:
|
|
return False
|
|
if row.owner_id != _actor(principal) and not _has_scope(principal, ADMIN_SCOPE):
|
|
raise PermissionError("Only the saved-view owner or an admin may delete it.")
|
|
session.delete(row)
|
|
session.flush()
|
|
return True
|
|
|
|
|
|
def upsert_schedule(
|
|
session: Session,
|
|
principal: object,
|
|
*,
|
|
schedule_id: str,
|
|
report_id: str,
|
|
report_revision: int,
|
|
name: str,
|
|
trigger_kind: str,
|
|
trigger_config: Mapping[str, object],
|
|
parameters: Mapping[str, object],
|
|
query: ReportQuery,
|
|
publication_target: Mapping[str, object],
|
|
enabled: bool,
|
|
next_run_at: datetime | None,
|
|
expected_revision: int | None,
|
|
) -> dict[str, object]:
|
|
_require_scope(principal, SCHEDULE_SCOPE)
|
|
report = get_definition(
|
|
session,
|
|
principal,
|
|
definition_kind="report",
|
|
definition_id=report_id,
|
|
revision=report_revision,
|
|
)
|
|
if report is None or report.status != "active":
|
|
raise ReportingOperationError("Schedules require an active report revision.")
|
|
normalized_next = _validate_trigger(
|
|
trigger_kind,
|
|
trigger_config,
|
|
next_run_at=next_run_at,
|
|
)
|
|
row = (
|
|
session.query(ReportingSchedule)
|
|
.filter(
|
|
ReportingSchedule.tenant_id == _tenant(principal),
|
|
ReportingSchedule.schedule_id == schedule_id,
|
|
)
|
|
.with_for_update()
|
|
.one_or_none()
|
|
)
|
|
if row is None:
|
|
if expected_revision is not None:
|
|
raise ReportingOperationError(
|
|
"Reporting schedule revision conflict: no schedule exists."
|
|
)
|
|
row = ReportingSchedule(
|
|
tenant_id=_tenant(principal),
|
|
schedule_id=_required(schedule_id, "Reporting schedule identifier", 36),
|
|
report_id=report_id,
|
|
report_revision=report_revision,
|
|
name=_required(name, "Reporting schedule name", 500),
|
|
revision=1,
|
|
trigger_kind=trigger_kind,
|
|
trigger_config=_json_value(trigger_config),
|
|
parameters=_json_value(parameters),
|
|
query=query.model_dump(mode="json"),
|
|
publication_target=_json_value(publication_target),
|
|
enabled=enabled,
|
|
next_run_at=normalized_next,
|
|
created_by=_actor(principal),
|
|
)
|
|
session.add(row)
|
|
else:
|
|
if expected_revision != row.revision:
|
|
raise ReportingOperationError(
|
|
"Reporting schedule revision conflict: the expected revision is stale."
|
|
)
|
|
row.report_id = report_id
|
|
row.report_revision = report_revision
|
|
row.name = _required(name, "Reporting schedule name", 500)
|
|
row.trigger_kind = trigger_kind
|
|
row.trigger_config = _json_value(trigger_config)
|
|
row.parameters = _json_value(parameters)
|
|
row.query = query.model_dump(mode="json")
|
|
row.publication_target = _json_value(publication_target)
|
|
row.enabled = enabled
|
|
row.next_run_at = normalized_next
|
|
row.revision += 1
|
|
session.flush()
|
|
return _schedule_payload(row)
|
|
|
|
|
|
def list_schedules(
|
|
session: Session,
|
|
principal: object,
|
|
*,
|
|
report_id: str | None = None,
|
|
) -> tuple[dict[str, object], ...]:
|
|
statement = session.query(ReportingSchedule).filter(
|
|
ReportingSchedule.tenant_id == _tenant(principal)
|
|
)
|
|
if report_id:
|
|
if (
|
|
get_definition(
|
|
session,
|
|
principal,
|
|
definition_kind="report",
|
|
definition_id=report_id,
|
|
)
|
|
is None
|
|
):
|
|
return ()
|
|
statement = statement.filter(ReportingSchedule.report_id == report_id)
|
|
return tuple(
|
|
_schedule_payload(row)
|
|
for row in statement.order_by(ReportingSchedule.name.asc()).all()
|
|
)
|
|
|
|
|
|
def dispatch_due_schedules(
|
|
session: Session,
|
|
principal: object,
|
|
*,
|
|
registry: object | None,
|
|
now: datetime | None,
|
|
limit: int,
|
|
) -> dict[str, object]:
|
|
_require_scope(principal, SCHEDULE_SCOPE)
|
|
current = now or utc_now()
|
|
_aware(current, "Reporting scheduler time")
|
|
rows = (
|
|
session.query(ReportingSchedule)
|
|
.filter(
|
|
ReportingSchedule.enabled.is_(True),
|
|
ReportingSchedule.next_run_at.is_not(None),
|
|
ReportingSchedule.next_run_at <= current,
|
|
)
|
|
.order_by(ReportingSchedule.next_run_at.asc())
|
|
.limit(max(1, min(limit, 100)))
|
|
.with_for_update(skip_locked=True)
|
|
.all()
|
|
)
|
|
succeeded = 0
|
|
failed = 0
|
|
execution_ids: list[str] = []
|
|
for row in rows:
|
|
scheduled_for = row.next_run_at or current
|
|
try:
|
|
execution = execute_report(
|
|
session,
|
|
principal,
|
|
registry=registry,
|
|
report_id=row.report_id,
|
|
report_revision=row.report_revision,
|
|
parameters=dict(row.parameters or {}),
|
|
query=ReportQuery.model_validate(row.query or {}),
|
|
idempotency_key=f"schedule:{row.schedule_id}:{scheduled_for.isoformat()}",
|
|
)
|
|
execution_id = str(execution["execution_id"])
|
|
execution_ids.append(execution_id)
|
|
target = dict(row.publication_target or {})
|
|
if target:
|
|
publish_execution(
|
|
session,
|
|
principal,
|
|
registry=registry,
|
|
execution_id=execution_id,
|
|
target_capability=str(target.get("target_capability") or ""),
|
|
target_ref=(
|
|
str(target["target_ref"])
|
|
if target.get("target_ref") is not None
|
|
else None
|
|
),
|
|
format=str(target.get("format") or "csv"),
|
|
idempotency_key=f"schedule-publication:{row.schedule_id}:{scheduled_for.isoformat()}",
|
|
options=dict(target.get("options") or {}),
|
|
)
|
|
succeeded += 1
|
|
row.last_execution_id = execution_id
|
|
except (ReportingExecutionFailure, ReportingOperationError, LookupError):
|
|
failed += 1
|
|
row.last_run_at = current
|
|
_advance_schedule(row, scheduled_for)
|
|
session.flush()
|
|
return {
|
|
"claimed": len(rows),
|
|
"succeeded": succeeded,
|
|
"failed": failed,
|
|
"execution_ids": execution_ids,
|
|
}
|
|
|
|
|
|
def publish_execution(
|
|
session: Session,
|
|
principal: object,
|
|
*,
|
|
registry: object | None,
|
|
execution_id: str,
|
|
target_capability: str,
|
|
target_ref: str | None,
|
|
format: str,
|
|
idempotency_key: str,
|
|
options: Mapping[str, object],
|
|
) -> dict[str, object]:
|
|
_require_scope(principal, PUBLISH_SCOPE)
|
|
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":
|
|
raise ReportingOperationError("Only successful report executions can publish.")
|
|
clean_capability = _required(
|
|
target_capability,
|
|
"Reporting publication target capability",
|
|
255,
|
|
)
|
|
clean_format = str(format).casefold()
|
|
if clean_format not in {"json", "csv", "xlsx", "html", "pdf"}:
|
|
raise ReportingOperationError("Unsupported Reporting publication format.")
|
|
clean_key = _required(idempotency_key, "Reporting publication idempotency key", 255)
|
|
existing = (
|
|
session.query(ReportingPublication)
|
|
.filter(
|
|
ReportingPublication.tenant_id == _tenant(principal),
|
|
ReportingPublication.idempotency_key == clean_key,
|
|
)
|
|
.one_or_none()
|
|
)
|
|
if existing is not None:
|
|
if (
|
|
existing.execution_id != execution_id
|
|
or existing.target_capability != clean_capability
|
|
or existing.target_ref != target_ref
|
|
or existing.format != clean_format
|
|
):
|
|
raise ReportingOperationError("Reporting publication idempotency conflict.")
|
|
return _publication_payload(existing)
|
|
provider = capability(registry, clean_capability)
|
|
if not isinstance(provider, ReportingPublicationTarget):
|
|
raise ReportingOperationError(
|
|
f"Reporting publication provider {clean_capability!r} is unavailable."
|
|
)
|
|
publication = ReportingPublication(
|
|
tenant_id=_tenant(principal),
|
|
publication_id=str(uuid.uuid4()),
|
|
execution_id=execution_id,
|
|
target_capability=clean_capability,
|
|
target_ref=target_ref,
|
|
format=clean_format,
|
|
status="running",
|
|
idempotency_key=clean_key,
|
|
)
|
|
session.add(publication)
|
|
session.flush()
|
|
try:
|
|
evidence = provider.publish_report(
|
|
session,
|
|
principal,
|
|
payload=ReportingPublicationPayload(
|
|
publication_id=publication.publication_id,
|
|
execution_id=execution_id,
|
|
tenant_id=_tenant(principal),
|
|
report_id=str(execution_payload["report_id"]),
|
|
report_revision=int(execution_payload["report_revision"]),
|
|
format=clean_format,
|
|
target_ref=target_ref,
|
|
rows=tuple(execution_payload["rows"]), # type: ignore[arg-type]
|
|
schema=tuple(execution_payload["schema"]), # type: ignore[arg-type]
|
|
output_hash=str(execution_payload["output_hash"]),
|
|
options=dict(options),
|
|
),
|
|
)
|
|
publication.status = "succeeded"
|
|
publication.evidence = _json_value(evidence)
|
|
publication.completed_at = utc_now()
|
|
except Exception as exc:
|
|
publication.status = "failed"
|
|
publication.error = str(exc)
|
|
publication.completed_at = utc_now()
|
|
session.flush()
|
|
raise ReportingOperationError(str(exc)) from exc
|
|
session.flush()
|
|
emit_platform_event(
|
|
session,
|
|
PlatformEvent(
|
|
type=f"reporting.publication.{publication.status}",
|
|
module_id="reporting",
|
|
payload={
|
|
"publication_id": publication.publication_id,
|
|
"execution_id": publication.execution_id,
|
|
"target_capability": publication.target_capability,
|
|
"target_ref": publication.target_ref,
|
|
"format": publication.format,
|
|
"evidence": dict(publication.evidence or {}),
|
|
},
|
|
actor=EventActorRef(type="account", id=_actor(principal)),
|
|
tenant=EventTenantRef(id=publication.tenant_id),
|
|
resource=EventObjectRef(
|
|
type="report_publication",
|
|
id=publication.publication_id,
|
|
),
|
|
classification="internal",
|
|
),
|
|
)
|
|
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,
|
|
registry=registry,
|
|
)
|
|
if payload is None:
|
|
raise LookupError("Reporting execution not found.")
|
|
if payload["status"] != "succeeded":
|
|
raise ReportingOperationError("Only successful executions can be exported.")
|
|
rows = tuple(payload["rows"]) # type: ignore[arg-type]
|
|
if format == "json":
|
|
content = json.dumps(
|
|
{"schema": payload["schema"], "rows": rows},
|
|
ensure_ascii=False,
|
|
separators=(",", ":"),
|
|
).encode("utf-8")
|
|
return content, "application/json", f"report-{execution_id}.json"
|
|
if format != "csv":
|
|
raise ReportingOperationError("Direct export supports CSV or JSON.")
|
|
fields = tuple(
|
|
dict.fromkeys(
|
|
str(key) for row in rows if isinstance(row, Mapping) for key in row
|
|
)
|
|
)
|
|
stream = StringIO(newline="")
|
|
writer = csv.DictWriter(stream, fieldnames=fields, extrasaction="ignore")
|
|
writer.writeheader()
|
|
for row in rows:
|
|
if isinstance(row, Mapping):
|
|
writer.writerow({key: _safe_csv_cell(row.get(key)) for key in fields})
|
|
return (
|
|
stream.getvalue().encode("utf-8-sig"),
|
|
"text/csv; charset=utf-8",
|
|
f"report-{execution_id}.csv",
|
|
)
|
|
|
|
|
|
def assess_import(
|
|
session: Session,
|
|
principal: object,
|
|
*,
|
|
source_system: str,
|
|
source_id: str,
|
|
metadata: Mapping[str, object],
|
|
accepted_approximations: list[str],
|
|
) -> dict[str, object]:
|
|
_require_scope(principal, IMPORT_SCOPE)
|
|
raw_features = metadata.get("features", [])
|
|
if not isinstance(raw_features, list):
|
|
raise ReportingOperationError("Import metadata features must be a list.")
|
|
features = tuple(dict.fromkeys(str(item) for item in raw_features))
|
|
exact_features = {
|
|
"dataset",
|
|
"dimension",
|
|
"hierarchy",
|
|
"measure",
|
|
"parameter",
|
|
"table",
|
|
"pivot",
|
|
"chart",
|
|
"quality_assertion",
|
|
"saved_view",
|
|
}
|
|
approximated_features = {
|
|
"provider_specific_format",
|
|
"dialect_function",
|
|
"dashboard_layout",
|
|
}
|
|
unsupported_features = {
|
|
"raw_sql",
|
|
"stored_procedure",
|
|
"runtime_script",
|
|
"implicit_authorization",
|
|
"unchecked_custom_function",
|
|
}
|
|
exact = sorted(set(features) & exact_features)
|
|
approximated = sorted(set(features) & approximated_features)
|
|
unsupported = sorted(
|
|
(set(features) & unsupported_features)
|
|
| (
|
|
set(features)
|
|
- exact_features
|
|
- approximated_features
|
|
- unsupported_features
|
|
)
|
|
)
|
|
accepted = sorted(set(accepted_approximations) & set(approximated))
|
|
pending = sorted(set(approximated) - set(accepted))
|
|
status = "ready" if not unsupported and not pending else "blocked"
|
|
mapping_report = {
|
|
"contract_version": "1",
|
|
"source_system": source_system,
|
|
"source_id": source_id,
|
|
"exact": exact,
|
|
"approximated": approximated,
|
|
"accepted_approximations": accepted,
|
|
"pending_approximations": pending,
|
|
"unsupported": unsupported,
|
|
"manual_bindings": list(metadata.get("manual_bindings", [])),
|
|
"provider_assumptions": list(metadata.get("provider_assumptions", [])),
|
|
"activation_allowed": status == "ready",
|
|
}
|
|
row = ReportingImportAssessment(
|
|
tenant_id=_tenant(principal),
|
|
assessment_id=str(uuid.uuid4()),
|
|
source_system=_required(source_system, "Import source system", 255),
|
|
source_id=_required(source_id, "Import source identifier", 500),
|
|
source_fingerprint=_sha256(metadata),
|
|
mapping_report=mapping_report,
|
|
status=status,
|
|
accepted_approximations=accepted,
|
|
assessed_by=_actor(principal),
|
|
)
|
|
session.add(row)
|
|
session.flush()
|
|
return _assessment_payload(row)
|
|
|
|
|
|
def list_import_assessments(
|
|
session: Session,
|
|
principal: object,
|
|
*,
|
|
limit: int = 100,
|
|
) -> tuple[dict[str, object], ...]:
|
|
_require_scope(principal, IMPORT_SCOPE)
|
|
rows = (
|
|
session.query(ReportingImportAssessment)
|
|
.filter(ReportingImportAssessment.tenant_id == _tenant(principal))
|
|
.order_by(ReportingImportAssessment.created_at.desc())
|
|
.limit(max(1, min(limit, 200)))
|
|
.all()
|
|
)
|
|
return tuple(_assessment_payload(row) for row in rows)
|
|
|
|
|
|
def _validate_saved_view_state(state: Mapping[str, object]) -> None:
|
|
query = state.get("query")
|
|
if query is not None:
|
|
if not isinstance(query, Mapping):
|
|
raise ReportingOperationError("Saved-view query must be an object.")
|
|
ReportQuery.model_validate(query)
|
|
if len(state) > 100:
|
|
raise ReportingOperationError("Saved-view state is limited to 100 entries.")
|
|
|
|
|
|
def _validate_trigger(
|
|
trigger_kind: str,
|
|
trigger_config: Mapping[str, object],
|
|
*,
|
|
next_run_at: datetime | None,
|
|
) -> datetime | None:
|
|
if trigger_kind not in {"scheduled", "interval"}:
|
|
raise ReportingOperationError("Unsupported Reporting schedule trigger.")
|
|
if next_run_at is not None:
|
|
_aware(next_run_at, "Reporting next_run_at")
|
|
if trigger_kind == "scheduled":
|
|
if next_run_at is None:
|
|
raise ReportingOperationError("Scheduled reports require next_run_at.")
|
|
return next_run_at
|
|
seconds = int(trigger_config.get("seconds", 0))
|
|
if not 60 <= seconds <= 31_536_000:
|
|
raise ReportingOperationError(
|
|
"Reporting intervals must be between 60 seconds and one year."
|
|
)
|
|
return next_run_at or utc_now() + timedelta(seconds=seconds)
|
|
|
|
|
|
def _advance_schedule(row: ReportingSchedule, scheduled_for: datetime) -> None:
|
|
if row.trigger_kind == "scheduled":
|
|
row.enabled = False
|
|
row.next_run_at = None
|
|
return
|
|
if scheduled_for.tzinfo is None or scheduled_for.utcoffset() is None:
|
|
scheduled_for = scheduled_for.replace(tzinfo=UTC)
|
|
seconds = int((row.trigger_config or {}).get("seconds", 0))
|
|
next_run = scheduled_for + timedelta(seconds=seconds)
|
|
now = utc_now()
|
|
while next_run <= now:
|
|
next_run += timedelta(seconds=seconds)
|
|
row.next_run_at = next_run
|
|
|
|
|
|
def _saved_view_payload(row: ReportingSavedView) -> dict[str, object]:
|
|
return {
|
|
"view_id": row.view_id,
|
|
"report_id": row.report_id,
|
|
"report_revision": row.report_revision,
|
|
"owner_kind": row.owner_kind,
|
|
"owner_id": row.owner_id,
|
|
"name": row.name,
|
|
"revision": row.revision,
|
|
"state": dict(row.state or {}),
|
|
"shared": row.shared,
|
|
"access": dict(row.access or {}),
|
|
"updated_at": _datetime_text(row.updated_at),
|
|
}
|
|
|
|
|
|
def _schedule_payload(row: ReportingSchedule) -> dict[str, object]:
|
|
return {
|
|
"schedule_id": row.schedule_id,
|
|
"report_id": row.report_id,
|
|
"report_revision": row.report_revision,
|
|
"name": row.name,
|
|
"revision": row.revision,
|
|
"trigger_kind": row.trigger_kind,
|
|
"trigger_config": dict(row.trigger_config or {}),
|
|
"parameters": dict(row.parameters or {}),
|
|
"query": dict(row.query or {}),
|
|
"publication_target": dict(row.publication_target or {}),
|
|
"enabled": row.enabled,
|
|
"next_run_at": _datetime_text(row.next_run_at),
|
|
"last_run_at": _datetime_text(row.last_run_at),
|
|
"last_execution_id": row.last_execution_id,
|
|
}
|
|
|
|
|
|
def _publication_payload(row: ReportingPublication) -> dict[str, object]:
|
|
return {
|
|
"publication_id": row.publication_id,
|
|
"execution_id": row.execution_id,
|
|
"target_capability": row.target_capability,
|
|
"target_ref": row.target_ref,
|
|
"format": row.format,
|
|
"status": row.status,
|
|
"evidence": dict(row.evidence or {}),
|
|
"error": row.error,
|
|
"completed_at": _datetime_text(row.completed_at),
|
|
}
|
|
|
|
|
|
def _assessment_payload(row: ReportingImportAssessment) -> dict[str, object]:
|
|
return {
|
|
"assessment_id": row.assessment_id,
|
|
"source_system": row.source_system,
|
|
"source_id": row.source_id,
|
|
"source_fingerprint": row.source_fingerprint,
|
|
"mapping_report": dict(row.mapping_report or {}),
|
|
"status": row.status,
|
|
"accepted_approximations": list(row.accepted_approximations or []),
|
|
"assessed_by": row.assessed_by,
|
|
"created_at": _datetime_text(row.created_at),
|
|
}
|
|
|
|
|
|
def _safe_csv_cell(value: object) -> object:
|
|
if isinstance(value, (dict, list, tuple)):
|
|
return json.dumps(value, ensure_ascii=False, separators=(",", ":"))
|
|
if isinstance(value, str) and value.startswith(("=", "+", "-", "@")):
|
|
return f"'{value}"
|
|
return value
|
|
|
|
|
|
def _has_scope(principal: object, scope: str) -> bool:
|
|
method = getattr(principal, "has", None)
|
|
if callable(method):
|
|
return bool(method(scope))
|
|
return scopes_grant_compatible(
|
|
frozenset(getattr(principal, "scopes", ()) or ()),
|
|
scope,
|
|
)
|
|
|
|
|
|
def _require_scope(principal: object, scope: str) -> None:
|
|
if not _has_scope(principal, scope):
|
|
raise PermissionError(f"Missing scope: {scope}")
|
|
|
|
|
|
def _tenant(principal: object) -> str:
|
|
tenant_id = str(getattr(principal, "tenant_id", "") or "").strip()
|
|
if not tenant_id:
|
|
raise ReportingOperationError(
|
|
"Reporting operations require a tenant-bound principal."
|
|
)
|
|
return tenant_id
|
|
|
|
|
|
def _actor(principal: object) -> str | None:
|
|
for value in (
|
|
getattr(principal, "account_id", None),
|
|
getattr(principal, "identity_id", None),
|
|
getattr(principal, "membership_id", None),
|
|
):
|
|
if str(value or "").strip():
|
|
return str(value)
|
|
return None
|
|
|
|
|
|
def _session(value: object) -> Session:
|
|
if not isinstance(value, Session):
|
|
raise TypeError("Reporting operations require a SQLAlchemy session.")
|
|
return value
|
|
|
|
|
|
def _required(value: object, label: str, maximum: int) -> str:
|
|
result = str(value or "").strip()
|
|
if not result:
|
|
raise ReportingOperationError(f"{label} is required.")
|
|
if len(result) > maximum:
|
|
raise ReportingOperationError(f"{label} is limited to {maximum} characters.")
|
|
return result
|
|
|
|
|
|
def _aware(value: datetime, label: str) -> None:
|
|
if value.tzinfo is None or value.utcoffset() is None:
|
|
raise ReportingOperationError(f"{label} must include a timezone.")
|
|
|
|
|
|
def _sha256(value: object) -> str:
|
|
return hashlib.sha256(
|
|
json.dumps(
|
|
_json_value(value),
|
|
sort_keys=True,
|
|
separators=(",", ":"),
|
|
ensure_ascii=True,
|
|
).encode("utf-8")
|
|
).hexdigest()
|
|
|
|
|
|
def _json_value(value: object) -> Any:
|
|
if isinstance(value, datetime):
|
|
return value.isoformat()
|
|
if hasattr(value, "model_dump"):
|
|
return _json_value(value.model_dump(mode="json"))
|
|
if isinstance(value, Mapping):
|
|
return {str(key): _json_value(item) for key, item in value.items()}
|
|
if isinstance(value, (list, tuple)):
|
|
return [_json_value(item) for item in value]
|
|
return value
|
|
|
|
|
|
def _datetime_text(value: datetime | None) -> str | None:
|
|
if value is None:
|
|
return None
|
|
if value.tzinfo is None:
|
|
value = value.replace(tzinfo=UTC)
|
|
return value.isoformat()
|
|
|
|
|
|
__all__ = [
|
|
"IMPORT_SCOPE",
|
|
"PUBLISH_SCOPE",
|
|
"SCHEDULE_SCOPE",
|
|
"ReportingOperationError",
|
|
"SqlReportingScheduler",
|
|
"assess_import",
|
|
"delete_saved_view",
|
|
"dispatch_due_schedules",
|
|
"export_execution",
|
|
"list_import_assessments",
|
|
"list_publications",
|
|
"list_saved_views",
|
|
"list_schedules",
|
|
"publish_execution",
|
|
"upsert_saved_view",
|
|
"upsert_schedule",
|
|
]
|