feat(reporting): add governed DSAR coverage
This commit is contained in:
@@ -0,0 +1,951 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Sequence
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime, timezone
|
||||
from typing import Any
|
||||
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from govoplan_core.core.dsar import (
|
||||
DsarErasureActionRef,
|
||||
DsarExecutionResultRef,
|
||||
DsarRecordRef,
|
||||
DsarSubjectRef,
|
||||
dsar_capability_name,
|
||||
)
|
||||
from govoplan_reporting.backend.db.models import (
|
||||
ReportingDefinitionGrant,
|
||||
ReportingDefinitionIdentity,
|
||||
ReportingDefinitionRevision,
|
||||
ReportingDrillContext,
|
||||
ReportingExecution,
|
||||
ReportingImportAssessment,
|
||||
ReportingProviderExecution,
|
||||
ReportingProviderExport,
|
||||
ReportingPublication,
|
||||
ReportingQualityResult,
|
||||
ReportingSavedView,
|
||||
ReportingSchedule,
|
||||
)
|
||||
|
||||
|
||||
REPORTING_DSAR_CAPABILITY = dsar_capability_name("reporting")
|
||||
_MAX_RECORDS = 5_000
|
||||
_CONFLICT = object()
|
||||
_DIRECT_ALIASES = {
|
||||
"definition_id": ("reporting.definition",),
|
||||
"revision_id": (
|
||||
"reporting.definition_revision",
|
||||
"reporting.revision",
|
||||
),
|
||||
"execution_id": ("reporting.execution",),
|
||||
"provider_execution_id": ("reporting.provider_execution",),
|
||||
"provider_export_id": ("reporting.provider_export",),
|
||||
"grant_id": ("reporting.definition_grant", "reporting.grant"),
|
||||
"saved_view_id": ("reporting.saved_view",),
|
||||
"schedule_id": ("reporting.schedule",),
|
||||
"publication_id": ("reporting.publication",),
|
||||
"drill_context_id": ("reporting.drill_context",),
|
||||
"quality_result_id": ("reporting.quality_result",),
|
||||
"import_assessment_id": ("reporting.import_assessment",),
|
||||
}
|
||||
_RESOURCE_MODELS = {
|
||||
"reporting_definition": ReportingDefinitionIdentity,
|
||||
"reporting_definition_revision": ReportingDefinitionRevision,
|
||||
"reporting_execution": ReportingExecution,
|
||||
"reporting_provider_execution": ReportingProviderExecution,
|
||||
"reporting_provider_export": ReportingProviderExport,
|
||||
"reporting_definition_grant": ReportingDefinitionGrant,
|
||||
"reporting_saved_view": ReportingSavedView,
|
||||
"reporting_schedule": ReportingSchedule,
|
||||
"reporting_publication": ReportingPublication,
|
||||
"reporting_drill_context": ReportingDrillContext,
|
||||
"reporting_quality_result": ReportingQualityResult,
|
||||
"reporting_import_assessment": ReportingImportAssessment,
|
||||
}
|
||||
_EXECUTABLE_KINDS = {
|
||||
"reporting_execution": "anonymize",
|
||||
"reporting_provider_execution": "anonymize",
|
||||
"reporting_provider_export": "anonymize",
|
||||
"reporting_publication": "anonymize",
|
||||
"reporting_definition_grant": "revoke",
|
||||
"reporting_saved_view": "delete",
|
||||
"reporting_drill_context": "delete",
|
||||
}
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class _Selectors:
|
||||
account_id: str | None
|
||||
identity_id: str | None
|
||||
membership_id: str | None
|
||||
direct: dict[str, str]
|
||||
|
||||
@property
|
||||
def actor_ids(self) -> tuple[str, ...]:
|
||||
return tuple(
|
||||
value
|
||||
for value in (self.account_id, self.identity_id, self.membership_id)
|
||||
if value
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class _Match:
|
||||
resource_type: str
|
||||
row: Any
|
||||
category: str
|
||||
|
||||
|
||||
class ReportingDsarProvider:
|
||||
provider_id = "reporting"
|
||||
module_id = "reporting"
|
||||
|
||||
def search_subject(
|
||||
self,
|
||||
session: object,
|
||||
*,
|
||||
tenant_id: str,
|
||||
subject: DsarSubjectRef,
|
||||
) -> Sequence[DsarRecordRef]:
|
||||
db = _session(session)
|
||||
selectors = _selectors(subject)
|
||||
if selectors is None or not (selectors.actor_ids or selectors.direct):
|
||||
return ()
|
||||
|
||||
direct = _direct_matches(db, tenant_id=tenant_id, selectors=selectors)
|
||||
if direct is None:
|
||||
return ()
|
||||
if direct:
|
||||
if selectors.actor_ids and not all(
|
||||
_correlates(
|
||||
db,
|
||||
tenant_id=tenant_id,
|
||||
match=match,
|
||||
actor_ids=selectors.actor_ids,
|
||||
)
|
||||
for match in direct
|
||||
):
|
||||
return ()
|
||||
matches = direct
|
||||
else:
|
||||
matches = _canonical_matches(
|
||||
db,
|
||||
tenant_id=tenant_id,
|
||||
actor_ids=selectors.actor_ids,
|
||||
)
|
||||
|
||||
records: list[DsarRecordRef] = []
|
||||
seen: set[tuple[str, str]] = set()
|
||||
for match in matches:
|
||||
key = (match.resource_type, str(match.row.id))
|
||||
if key in seen:
|
||||
continue
|
||||
if len(records) >= _MAX_RECORDS:
|
||||
raise ValueError(
|
||||
"Reporting DSAR result limit exceeded; narrow the selectors."
|
||||
)
|
||||
seen.add(key)
|
||||
records.append(_record(match))
|
||||
return tuple(records)
|
||||
|
||||
def plan_erasure(
|
||||
self,
|
||||
session: object,
|
||||
*,
|
||||
tenant_id: str,
|
||||
subject: DsarSubjectRef,
|
||||
records: Sequence[DsarRecordRef],
|
||||
) -> Sequence[DsarErasureActionRef]:
|
||||
del tenant_id
|
||||
_session(session)
|
||||
if _selectors(subject) is None:
|
||||
raise ValueError("Reporting DSAR subject selectors conflict.")
|
||||
actions: list[DsarErasureActionRef] = []
|
||||
for record in records:
|
||||
_validate_record(record)
|
||||
kind = _planned_kind(record)
|
||||
executable = kind in {"delete", "anonymize", "revoke"}
|
||||
actions.append(
|
||||
DsarErasureActionRef(
|
||||
action_id=(
|
||||
f"reporting:{kind}:{record.resource_type}:{record.resource_id}"
|
||||
),
|
||||
provider_id=self.provider_id,
|
||||
module_id=self.module_id,
|
||||
kind=kind,
|
||||
resource_type=record.resource_type,
|
||||
resource_id=record.resource_id,
|
||||
title=(
|
||||
f"Minimize {record.title}"
|
||||
if kind == "anonymize"
|
||||
else f"{kind.replace('_', ' ').title()} {record.title}"
|
||||
),
|
||||
rationale=_rationale(record, kind=kind),
|
||||
executable=executable,
|
||||
irreversible=kind in {"delete", "anonymize"},
|
||||
metadata={"record_category": record.category},
|
||||
)
|
||||
)
|
||||
return tuple(actions)
|
||||
|
||||
def execute_erasure(
|
||||
self,
|
||||
session: object,
|
||||
*,
|
||||
tenant_id: str,
|
||||
subject: DsarSubjectRef,
|
||||
actions: Sequence[DsarErasureActionRef],
|
||||
request_id: str,
|
||||
) -> Sequence[DsarExecutionResultRef]:
|
||||
db = _session(session)
|
||||
selectors = _selectors(subject)
|
||||
if selectors is None:
|
||||
raise ValueError("Reporting DSAR subject selectors conflict.")
|
||||
results: list[DsarExecutionResultRef] = []
|
||||
for action in actions:
|
||||
_validate_action(action)
|
||||
if not action.executable:
|
||||
results.append(
|
||||
DsarExecutionResultRef(
|
||||
action_id=action.action_id,
|
||||
status="blocked",
|
||||
summary=(
|
||||
"Review institutional reporting evidence, shared "
|
||||
"configuration, retention, and third-party impact."
|
||||
),
|
||||
evidence={"request_id": request_id},
|
||||
)
|
||||
)
|
||||
continue
|
||||
|
||||
model = _RESOURCE_MODELS[action.resource_type]
|
||||
row = (
|
||||
db.query(model)
|
||||
.filter(model.tenant_id == tenant_id, model.id == action.resource_id)
|
||||
.with_for_update()
|
||||
.one_or_none()
|
||||
)
|
||||
if row is None:
|
||||
status = "unchanged"
|
||||
summary = "Reporting row was already absent or minimized."
|
||||
else:
|
||||
match = _Match(action.resource_type, row, "execution")
|
||||
if not (
|
||||
_directly_targets(selectors, match)
|
||||
or _correlates(
|
||||
db,
|
||||
tenant_id=tenant_id,
|
||||
match=match,
|
||||
actor_ids=selectors.actor_ids,
|
||||
)
|
||||
):
|
||||
raise ValueError(
|
||||
"Reporting DSAR action is not corroborated by the subject."
|
||||
)
|
||||
status, summary = _execute_action(
|
||||
db,
|
||||
row=row,
|
||||
resource_type=action.resource_type,
|
||||
kind=action.kind,
|
||||
)
|
||||
results.append(
|
||||
DsarExecutionResultRef(
|
||||
action_id=action.action_id,
|
||||
status=status,
|
||||
summary=summary,
|
||||
evidence={"request_id": request_id},
|
||||
)
|
||||
)
|
||||
return tuple(results)
|
||||
|
||||
|
||||
def _direct_matches(
|
||||
session: Session,
|
||||
*,
|
||||
tenant_id: str,
|
||||
selectors: _Selectors,
|
||||
) -> list[_Match] | None:
|
||||
matches: list[_Match] = []
|
||||
for selector, value in selectors.direct.items():
|
||||
current: list[_Match]
|
||||
if selector == "definition_id":
|
||||
identities = _rows(
|
||||
session,
|
||||
ReportingDefinitionIdentity,
|
||||
tenant_id=tenant_id,
|
||||
field="definition_id",
|
||||
value=value,
|
||||
)
|
||||
revisions = _rows(
|
||||
session,
|
||||
ReportingDefinitionRevision,
|
||||
tenant_id=tenant_id,
|
||||
field="definition_id",
|
||||
value=value,
|
||||
)
|
||||
current = [
|
||||
*(
|
||||
_Match("reporting_definition", row, "reporting_configuration")
|
||||
for row in identities
|
||||
),
|
||||
*(
|
||||
_Match(
|
||||
"reporting_definition_revision",
|
||||
row,
|
||||
"reporting_configuration",
|
||||
)
|
||||
for row in revisions
|
||||
),
|
||||
]
|
||||
else:
|
||||
model, field, resource_type = {
|
||||
"revision_id": (
|
||||
ReportingDefinitionRevision,
|
||||
"id",
|
||||
"reporting_definition_revision",
|
||||
),
|
||||
"execution_id": (
|
||||
ReportingExecution,
|
||||
"execution_id",
|
||||
"reporting_execution",
|
||||
),
|
||||
"provider_execution_id": (
|
||||
ReportingProviderExecution,
|
||||
"execution_id",
|
||||
"reporting_provider_execution",
|
||||
),
|
||||
"provider_export_id": (
|
||||
ReportingProviderExport,
|
||||
"export_id",
|
||||
"reporting_provider_export",
|
||||
),
|
||||
"grant_id": (
|
||||
ReportingDefinitionGrant,
|
||||
"id",
|
||||
"reporting_definition_grant",
|
||||
),
|
||||
"saved_view_id": (
|
||||
ReportingSavedView,
|
||||
"view_id",
|
||||
"reporting_saved_view",
|
||||
),
|
||||
"schedule_id": (
|
||||
ReportingSchedule,
|
||||
"schedule_id",
|
||||
"reporting_schedule",
|
||||
),
|
||||
"publication_id": (
|
||||
ReportingPublication,
|
||||
"publication_id",
|
||||
"reporting_publication",
|
||||
),
|
||||
"drill_context_id": (
|
||||
ReportingDrillContext,
|
||||
"drill_context_id",
|
||||
"reporting_drill_context",
|
||||
),
|
||||
"quality_result_id": (
|
||||
ReportingQualityResult,
|
||||
"result_id",
|
||||
"reporting_quality_result",
|
||||
),
|
||||
"import_assessment_id": (
|
||||
ReportingImportAssessment,
|
||||
"assessment_id",
|
||||
"reporting_import_assessment",
|
||||
),
|
||||
}[selector]
|
||||
current = [
|
||||
_Match(resource_type, row, _direct_category(resource_type, row))
|
||||
for row in _rows(
|
||||
session,
|
||||
model,
|
||||
tenant_id=tenant_id,
|
||||
field=field,
|
||||
value=value,
|
||||
)
|
||||
]
|
||||
if not current:
|
||||
return None
|
||||
matches.extend(current)
|
||||
if len(matches) > _MAX_RECORDS:
|
||||
raise ValueError(
|
||||
"Reporting DSAR result limit exceeded; narrow the selectors."
|
||||
)
|
||||
return matches
|
||||
|
||||
|
||||
def _canonical_matches(
|
||||
session: Session,
|
||||
*,
|
||||
tenant_id: str,
|
||||
actor_ids: tuple[str, ...],
|
||||
) -> list[_Match]:
|
||||
if not actor_ids:
|
||||
return []
|
||||
specs = (
|
||||
(
|
||||
ReportingDefinitionIdentity,
|
||||
"created_by",
|
||||
"reporting_definition",
|
||||
"reporting_operator_attribution",
|
||||
),
|
||||
(
|
||||
ReportingDefinitionRevision,
|
||||
"changed_by",
|
||||
"reporting_definition_revision",
|
||||
"reporting_operator_attribution",
|
||||
),
|
||||
(
|
||||
ReportingExecution,
|
||||
"actor_id",
|
||||
"reporting_execution",
|
||||
"reporting_operator_attribution",
|
||||
),
|
||||
(
|
||||
ReportingProviderExecution,
|
||||
"actor_id",
|
||||
"reporting_provider_execution",
|
||||
"reporting_operator_attribution",
|
||||
),
|
||||
(
|
||||
ReportingProviderExport,
|
||||
"actor_id",
|
||||
"reporting_provider_export",
|
||||
"reporting_operator_attribution",
|
||||
),
|
||||
(
|
||||
ReportingSavedView,
|
||||
"owner_id",
|
||||
"reporting_saved_view",
|
||||
"subject_owned_reporting_view",
|
||||
),
|
||||
(
|
||||
ReportingSchedule,
|
||||
"created_by",
|
||||
"reporting_schedule",
|
||||
"reporting_operator_attribution",
|
||||
),
|
||||
(
|
||||
ReportingDrillContext,
|
||||
"actor_id",
|
||||
"reporting_drill_context",
|
||||
"subject_owned_drill_context",
|
||||
),
|
||||
(
|
||||
ReportingQualityResult,
|
||||
"actor_id",
|
||||
"reporting_quality_result",
|
||||
"reporting_operator_attribution",
|
||||
),
|
||||
(
|
||||
ReportingImportAssessment,
|
||||
"assessed_by",
|
||||
"reporting_import_assessment",
|
||||
"reporting_operator_attribution",
|
||||
),
|
||||
(
|
||||
ReportingDefinitionGrant,
|
||||
"subject_id",
|
||||
"reporting_definition_grant",
|
||||
"subject_access_grant",
|
||||
),
|
||||
)
|
||||
matches: list[_Match] = []
|
||||
for model, field, resource_type, category in specs:
|
||||
rows = (
|
||||
session.query(model)
|
||||
.filter(
|
||||
model.tenant_id == tenant_id,
|
||||
getattr(model, field).in_(actor_ids),
|
||||
)
|
||||
.order_by(model.id)
|
||||
.limit(_MAX_RECORDS + 1)
|
||||
.all()
|
||||
)
|
||||
matches.extend(
|
||||
_Match(
|
||||
resource_type,
|
||||
row,
|
||||
_direct_category(resource_type, row)
|
||||
if resource_type == "reporting_saved_view"
|
||||
else category,
|
||||
)
|
||||
for row in rows
|
||||
)
|
||||
if len(matches) > _MAX_RECORDS:
|
||||
raise ValueError(
|
||||
"Reporting DSAR result limit exceeded; narrow the selectors."
|
||||
)
|
||||
return matches
|
||||
|
||||
|
||||
def _rows(
|
||||
session: Session,
|
||||
model: Any,
|
||||
*,
|
||||
tenant_id: str,
|
||||
field: str,
|
||||
value: str,
|
||||
) -> list[Any]:
|
||||
return (
|
||||
session.query(model)
|
||||
.filter(model.tenant_id == tenant_id, getattr(model, field) == value)
|
||||
.order_by(model.id)
|
||||
.limit(_MAX_RECORDS + 1)
|
||||
.all()
|
||||
)
|
||||
|
||||
|
||||
def _correlates(
|
||||
session: Session,
|
||||
*,
|
||||
tenant_id: str,
|
||||
match: _Match,
|
||||
actor_ids: tuple[str, ...],
|
||||
) -> bool:
|
||||
if not actor_ids:
|
||||
return False
|
||||
row = match.row
|
||||
field = {
|
||||
"reporting_definition": "created_by",
|
||||
"reporting_definition_revision": "changed_by",
|
||||
"reporting_execution": "actor_id",
|
||||
"reporting_provider_execution": "actor_id",
|
||||
"reporting_provider_export": "actor_id",
|
||||
"reporting_definition_grant": "subject_id",
|
||||
"reporting_saved_view": "owner_id",
|
||||
"reporting_schedule": "created_by",
|
||||
"reporting_drill_context": "actor_id",
|
||||
"reporting_quality_result": "actor_id",
|
||||
"reporting_import_assessment": "assessed_by",
|
||||
}.get(match.resource_type)
|
||||
if field and str(getattr(row, field, "") or "") in actor_ids:
|
||||
return True
|
||||
if match.resource_type == "reporting_definition_revision":
|
||||
identity = session.get(ReportingDefinitionIdentity, row.identity_id)
|
||||
return bool(
|
||||
identity
|
||||
and identity.tenant_id == tenant_id
|
||||
and identity.created_by in actor_ids
|
||||
)
|
||||
if match.resource_type == "reporting_provider_export":
|
||||
execution = session.get(
|
||||
ReportingProviderExecution,
|
||||
row.provider_execution_id,
|
||||
)
|
||||
return bool(
|
||||
execution
|
||||
and execution.tenant_id == tenant_id
|
||||
and execution.actor_id in actor_ids
|
||||
)
|
||||
if match.resource_type == "reporting_publication":
|
||||
execution = (
|
||||
session.query(ReportingExecution)
|
||||
.filter(
|
||||
ReportingExecution.tenant_id == tenant_id,
|
||||
ReportingExecution.execution_id == row.execution_id,
|
||||
)
|
||||
.one_or_none()
|
||||
)
|
||||
return bool(execution and execution.actor_id in actor_ids)
|
||||
return False
|
||||
|
||||
|
||||
def _directly_targets(selectors: _Selectors, match: _Match) -> bool:
|
||||
row = match.row
|
||||
selector, field = {
|
||||
"reporting_definition": ("definition_id", "definition_id"),
|
||||
"reporting_definition_revision": ("revision_id", "id"),
|
||||
"reporting_execution": ("execution_id", "execution_id"),
|
||||
"reporting_provider_execution": (
|
||||
"provider_execution_id",
|
||||
"execution_id",
|
||||
),
|
||||
"reporting_provider_export": ("provider_export_id", "export_id"),
|
||||
"reporting_definition_grant": ("grant_id", "id"),
|
||||
"reporting_saved_view": ("saved_view_id", "view_id"),
|
||||
"reporting_schedule": ("schedule_id", "schedule_id"),
|
||||
"reporting_publication": ("publication_id", "publication_id"),
|
||||
"reporting_drill_context": ("drill_context_id", "drill_context_id"),
|
||||
"reporting_quality_result": ("quality_result_id", "result_id"),
|
||||
"reporting_import_assessment": (
|
||||
"import_assessment_id",
|
||||
"assessment_id",
|
||||
),
|
||||
}[match.resource_type]
|
||||
value = getattr(row, field)
|
||||
if selectors.direct.get(selector) == str(value):
|
||||
return True
|
||||
return (
|
||||
match.resource_type == "reporting_definition_revision"
|
||||
and selectors.direct.get("definition_id") == row.definition_id
|
||||
)
|
||||
|
||||
|
||||
def _direct_category(resource_type: str, row: Any) -> str:
|
||||
if resource_type == "reporting_saved_view":
|
||||
return "shared_reporting_view" if row.shared else "subject_owned_reporting_view"
|
||||
return {
|
||||
"reporting_execution": "derived_report_result",
|
||||
"reporting_provider_execution": "derived_provider_report_result",
|
||||
"reporting_provider_export": "derived_provider_report_export",
|
||||
"reporting_publication": "derived_report_publication",
|
||||
"reporting_definition_grant": "subject_access_grant",
|
||||
"reporting_drill_context": "subject_owned_drill_context",
|
||||
}.get(resource_type, "reporting_configuration")
|
||||
|
||||
|
||||
def _record(match: _Match) -> DsarRecordRef:
|
||||
row = match.row
|
||||
data = _record_data(match.resource_type, row)
|
||||
immutable = match.category == "reporting_operator_attribution"
|
||||
return DsarRecordRef(
|
||||
provider_id="reporting",
|
||||
module_id="reporting",
|
||||
resource_type=match.resource_type,
|
||||
resource_id=str(row.id),
|
||||
category=match.category,
|
||||
title=_title(match.resource_type),
|
||||
data={key: value for key, value in data.items() if value is not None},
|
||||
observed_at=_observed_at(row),
|
||||
immutable_evidence=immutable,
|
||||
retention_reason=(
|
||||
"Institutional reporting activity remains attributable for "
|
||||
"governance and audit review."
|
||||
if immutable
|
||||
else None
|
||||
),
|
||||
source_path="/reports",
|
||||
)
|
||||
|
||||
|
||||
def _record_data(resource_type: str, row: Any) -> dict[str, object]:
|
||||
if resource_type == "reporting_definition":
|
||||
return {
|
||||
"definition_kind": row.definition_kind,
|
||||
"definition_id": row.definition_id,
|
||||
"definition_key": row.definition_key,
|
||||
"created_at": _iso(row.created_at),
|
||||
}
|
||||
if resource_type == "reporting_definition_revision":
|
||||
return {
|
||||
"definition_kind": row.definition_kind,
|
||||
"definition_id": row.definition_id,
|
||||
"revision": row.revision,
|
||||
"status": row.status,
|
||||
"visibility": row.visibility,
|
||||
"recorded_at": _iso(row.recorded_at),
|
||||
"superseded_at": _iso(row.superseded_at),
|
||||
}
|
||||
if resource_type == "reporting_execution":
|
||||
return {
|
||||
"execution_id": row.execution_id,
|
||||
"report_id": row.report_id,
|
||||
"report_revision": row.report_revision,
|
||||
"semantic_model_id": row.semantic_model_id,
|
||||
"semantic_model_revision": row.semantic_model_revision,
|
||||
"dataset_id": row.dataset_id,
|
||||
"dataset_revision": row.dataset_revision,
|
||||
"status": row.status,
|
||||
"total_rows": row.total_rows,
|
||||
"truncated": row.truncated,
|
||||
"has_retained_result": bool(row.result_rows),
|
||||
"started_at": _iso(row.started_at),
|
||||
"finished_at": _iso(row.finished_at),
|
||||
}
|
||||
if resource_type == "reporting_provider_execution":
|
||||
return {
|
||||
"execution_id": row.execution_id,
|
||||
"provider_id": row.provider_id,
|
||||
"report_id": row.report_id,
|
||||
"report_revision": row.report_revision,
|
||||
"contract_version": row.contract_version,
|
||||
"privacy_transforms": list(row.privacy_transforms or []),
|
||||
"retention_class": row.retention_class,
|
||||
"retention_days": row.retention_days,
|
||||
"expires_at": _iso(row.expires_at),
|
||||
"retention_redacted_at": _iso(row.retention_redacted_at),
|
||||
"has_retained_result": bool(row.result_payload),
|
||||
"generated_at": _iso(row.generated_at),
|
||||
}
|
||||
if resource_type == "reporting_provider_export":
|
||||
return {
|
||||
"export_id": row.export_id,
|
||||
"execution_id": row.execution_id,
|
||||
"format": row.format,
|
||||
"exported_at": _iso(row.exported_at),
|
||||
}
|
||||
if resource_type == "reporting_definition_grant":
|
||||
return {
|
||||
"definition_kind": row.definition_kind,
|
||||
"definition_id": row.definition_id,
|
||||
"subject_kind": row.subject_kind,
|
||||
"permissions": list(row.permissions or []),
|
||||
"active": row.active,
|
||||
"source_revision": row.source_revision,
|
||||
}
|
||||
if resource_type == "reporting_saved_view":
|
||||
return {
|
||||
"view_id": row.view_id,
|
||||
"report_id": row.report_id,
|
||||
"report_revision": row.report_revision,
|
||||
"owner_kind": row.owner_kind,
|
||||
"revision": row.revision,
|
||||
"shared": row.shared,
|
||||
"created_at": _iso(row.created_at),
|
||||
"updated_at": _iso(row.updated_at),
|
||||
}
|
||||
if resource_type == "reporting_schedule":
|
||||
return {
|
||||
"schedule_id": row.schedule_id,
|
||||
"report_id": row.report_id,
|
||||
"report_revision": row.report_revision,
|
||||
"revision": row.revision,
|
||||
"trigger_kind": row.trigger_kind,
|
||||
"enabled": row.enabled,
|
||||
"next_run_at": _iso(row.next_run_at),
|
||||
"last_run_at": _iso(row.last_run_at),
|
||||
}
|
||||
if resource_type == "reporting_publication":
|
||||
return {
|
||||
"publication_id": row.publication_id,
|
||||
"execution_id": row.execution_id,
|
||||
"target_capability": row.target_capability,
|
||||
"format": row.format,
|
||||
"status": row.status,
|
||||
"completed_at": _iso(row.completed_at),
|
||||
"created_at": _iso(row.created_at),
|
||||
}
|
||||
if resource_type == "reporting_drill_context":
|
||||
return {
|
||||
"drill_context_id": row.drill_context_id,
|
||||
"execution_id": row.execution_id,
|
||||
"expires_at": _iso(row.expires_at),
|
||||
"last_accessed_at": _iso(row.last_accessed_at),
|
||||
}
|
||||
if resource_type == "reporting_quality_result":
|
||||
return {
|
||||
"result_id": row.result_id,
|
||||
"quality_plan_id": row.quality_plan_id,
|
||||
"quality_plan_revision": row.quality_plan_revision,
|
||||
"dataset_id": row.dataset_id,
|
||||
"dataset_revision": row.dataset_revision,
|
||||
"status": row.status,
|
||||
"evaluated_at": _iso(row.evaluated_at),
|
||||
}
|
||||
return {
|
||||
"assessment_id": row.assessment_id,
|
||||
"source_system": row.source_system,
|
||||
"status": row.status,
|
||||
"created_at": _iso(row.created_at),
|
||||
}
|
||||
|
||||
|
||||
def _planned_kind(record: DsarRecordRef) -> str:
|
||||
if record.category == "reporting_operator_attribution":
|
||||
return "retain"
|
||||
if record.category in {"reporting_configuration", "shared_reporting_view"}:
|
||||
return "manual_review"
|
||||
return _EXECUTABLE_KINDS.get(record.resource_type, "manual_review")
|
||||
|
||||
|
||||
def _rationale(record: DsarRecordRef, *, kind: str) -> str:
|
||||
if kind == "delete":
|
||||
return "Remove subject-owned, non-authoritative Reporting workspace state."
|
||||
if kind == "anonymize":
|
||||
return (
|
||||
"Clear retained result or delivery detail while preserving hashes and "
|
||||
"minimal institutional execution evidence."
|
||||
)
|
||||
if kind == "revoke":
|
||||
return "Disable the subject-specific Reporting access relationship."
|
||||
if kind == "retain":
|
||||
return record.retention_reason or "Retain institutional attribution evidence."
|
||||
return (
|
||||
"An authorized Reporting owner must review shared configuration, "
|
||||
"dependencies, legal retention, and third-party impact."
|
||||
)
|
||||
|
||||
|
||||
def _execute_action(
|
||||
session: Session,
|
||||
*,
|
||||
row: Any,
|
||||
resource_type: str,
|
||||
kind: str,
|
||||
) -> tuple[str, str]:
|
||||
expected = _EXECUTABLE_KINDS.get(resource_type)
|
||||
if expected != kind:
|
||||
raise ValueError("Reporting DSAR executable action is not supported.")
|
||||
if resource_type == "reporting_saved_view":
|
||||
if row.shared:
|
||||
raise ValueError("Shared Reporting views require manual review.")
|
||||
session.delete(row)
|
||||
session.flush()
|
||||
return "executed", "Subject-owned Reporting view removed."
|
||||
if resource_type == "reporting_drill_context":
|
||||
session.delete(row)
|
||||
session.flush()
|
||||
return "executed", "Ephemeral Reporting drill context removed."
|
||||
if resource_type == "reporting_definition_grant":
|
||||
if not row.active:
|
||||
return "unchanged", "Reporting access grant was already inactive."
|
||||
row.active = False
|
||||
session.flush()
|
||||
return "executed", "Subject-specific Reporting access grant revoked."
|
||||
if resource_type == "reporting_execution":
|
||||
fields = {
|
||||
"parameters": {},
|
||||
"query": {},
|
||||
"source_fingerprints": [],
|
||||
"result_rows": [],
|
||||
"diagnostics": [],
|
||||
"provenance": {},
|
||||
}
|
||||
changed = _replace_fields(row, fields)
|
||||
elif resource_type == "reporting_provider_execution":
|
||||
fields = {
|
||||
"purpose": "Redacted by data-subject request.",
|
||||
"audience_scope": {},
|
||||
"parameters": {},
|
||||
"result_payload": {},
|
||||
"source_revisions": [],
|
||||
"effective_scope": {},
|
||||
"provenance": {},
|
||||
"governance_provenance": {},
|
||||
}
|
||||
changed = _replace_fields(row, fields)
|
||||
if row.retention_redacted_at is None:
|
||||
row.retention_redacted_at = datetime.now(timezone.utc)
|
||||
changed = True
|
||||
elif resource_type == "reporting_provider_export":
|
||||
changed = _replace_fields(
|
||||
row,
|
||||
{
|
||||
"purpose": "Redacted by data-subject request.",
|
||||
"audience_scope": {},
|
||||
},
|
||||
)
|
||||
else:
|
||||
changed = _replace_fields(
|
||||
row,
|
||||
{"target_ref": None, "evidence": {}, "error": None},
|
||||
)
|
||||
if changed:
|
||||
session.flush()
|
||||
return "executed", "Retained Reporting detail minimized; hashes remain."
|
||||
return "unchanged", "Retained Reporting detail was already minimized."
|
||||
|
||||
|
||||
def _replace_fields(row: Any, values: dict[str, object]) -> bool:
|
||||
changed = False
|
||||
for field, value in values.items():
|
||||
if getattr(row, field) != value:
|
||||
setattr(row, field, value)
|
||||
changed = True
|
||||
return changed
|
||||
|
||||
|
||||
def _selectors(subject: DsarSubjectRef) -> _Selectors | None:
|
||||
references = subject.external_references
|
||||
account_id = _coalesce(
|
||||
subject.account_id,
|
||||
references.get("reporting.account"),
|
||||
references.get("access.account"),
|
||||
)
|
||||
identity_id = _coalesce(
|
||||
subject.identity_id,
|
||||
references.get("reporting.identity"),
|
||||
references.get("identity.id"),
|
||||
)
|
||||
membership_id = _coalesce(
|
||||
subject.membership_id,
|
||||
references.get("reporting.membership"),
|
||||
references.get("tenancy.membership"),
|
||||
)
|
||||
direct: dict[str, str] = {}
|
||||
for selector, aliases in _DIRECT_ALIASES.items():
|
||||
value = _coalesce(*(references.get(alias) for alias in aliases))
|
||||
if value is _CONFLICT:
|
||||
return None
|
||||
if value:
|
||||
direct[selector] = str(value)
|
||||
if _CONFLICT in {account_id, identity_id, membership_id}:
|
||||
return None
|
||||
return _Selectors(
|
||||
account_id=_optional(account_id),
|
||||
identity_id=_optional(identity_id),
|
||||
membership_id=_optional(membership_id),
|
||||
direct=direct,
|
||||
)
|
||||
|
||||
|
||||
def _coalesce(*values: str | None) -> str | None | object:
|
||||
normalized = {str(value).strip() for value in values if str(value or "").strip()}
|
||||
if len(normalized) > 1:
|
||||
return _CONFLICT
|
||||
return next(iter(normalized), None)
|
||||
|
||||
|
||||
def _optional(value: object) -> str | None:
|
||||
return value if isinstance(value, str) and value else None
|
||||
|
||||
|
||||
def _title(resource_type: str) -> str:
|
||||
return resource_type.removeprefix("reporting_").replace("_", " ").title()
|
||||
|
||||
|
||||
def _observed_at(row: Any) -> datetime | None:
|
||||
for field in (
|
||||
"generated_at",
|
||||
"exported_at",
|
||||
"evaluated_at",
|
||||
"recorded_at",
|
||||
"started_at",
|
||||
"completed_at",
|
||||
"updated_at",
|
||||
"created_at",
|
||||
):
|
||||
value = getattr(row, field, None)
|
||||
if isinstance(value, datetime):
|
||||
return _aware(value)
|
||||
return None
|
||||
|
||||
|
||||
def _iso(value: datetime | None) -> str | None:
|
||||
aware = _aware(value)
|
||||
return aware.isoformat() if aware else None
|
||||
|
||||
|
||||
def _aware(value: datetime | None) -> datetime | None:
|
||||
if value is None or value.tzinfo is not None:
|
||||
return value
|
||||
return value.replace(tzinfo=timezone.utc)
|
||||
|
||||
|
||||
def _session(value: object) -> Session:
|
||||
if not isinstance(value, Session):
|
||||
raise TypeError("Reporting DSAR requires a SQLAlchemy Session.")
|
||||
return value
|
||||
|
||||
|
||||
def _validate_record(record: DsarRecordRef) -> None:
|
||||
if record.provider_id != "reporting" or record.module_id != "reporting":
|
||||
raise ValueError("Reporting DSAR cannot plan a foreign provider record.")
|
||||
if record.resource_type not in _RESOURCE_MODELS or not record.resource_id:
|
||||
raise ValueError("Reporting DSAR record identity is invalid.")
|
||||
|
||||
|
||||
def _validate_action(action: DsarErasureActionRef) -> None:
|
||||
if action.provider_id != "reporting" or action.module_id != "reporting":
|
||||
raise ValueError("Reporting DSAR cannot execute a foreign provider action.")
|
||||
if action.resource_type not in _RESOURCE_MODELS or not action.action_id.startswith(
|
||||
"reporting:"
|
||||
):
|
||||
raise ValueError("Reporting DSAR action identity is invalid.")
|
||||
|
||||
|
||||
__all__ = ["REPORTING_DSAR_CAPABILITY", "ReportingDsarProvider"]
|
||||
Reference in New Issue
Block a user