288 lines
11 KiB
Python
288 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
from collections.abc import Mapping, Sequence
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from govoplan_core.core.dsar import (
|
|
DsarErasureActionRef,
|
|
DsarExecutionResultRef,
|
|
DsarRecordRef,
|
|
DsarSubjectRef,
|
|
dsar_capability_name,
|
|
)
|
|
from govoplan_tickets.backend.db.models import Ticket, TicketComment, TicketHistory
|
|
|
|
|
|
TICKETS_DSAR_CAPABILITY = dsar_capability_name("tickets")
|
|
_MAX_RECORDS = 5_000
|
|
_CONFLICT = object()
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class _Selectors:
|
|
subject_ids: tuple[str, ...]
|
|
ticket_id: str | None
|
|
|
|
|
|
class TicketsDsarProvider:
|
|
provider_id = "tickets"
|
|
module_id = "tickets"
|
|
|
|
def search_subject(
|
|
self,
|
|
session: object,
|
|
*,
|
|
tenant_id: str,
|
|
subject: DsarSubjectRef,
|
|
) -> Sequence[DsarRecordRef]:
|
|
db = _session(session)
|
|
selectors = _selectors(subject)
|
|
if selectors is None:
|
|
return ()
|
|
query = db.query(Ticket).filter(Ticket.tenant_id == tenant_id)
|
|
if selectors.ticket_id:
|
|
query = query.filter(Ticket.id == selectors.ticket_id)
|
|
rows = query.limit(_MAX_RECORDS + 1).all()
|
|
if len(rows) > _MAX_RECORDS:
|
|
raise ValueError("Tickets DSAR result limit exceeded; narrow the selectors.")
|
|
records: list[DsarRecordRef] = []
|
|
for row in rows:
|
|
roles = _subject_roles(row, selectors.subject_ids)
|
|
activities = _activities(db, row, selectors.subject_ids)
|
|
comments = _comments(db, row, selectors.subject_ids)
|
|
if roles or comments:
|
|
records.append(_participation_record(row, roles, comments, activities))
|
|
elif activities:
|
|
records.append(_actor_record(row, activities))
|
|
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("Tickets DSAR subject selectors conflict.")
|
|
actions = []
|
|
for record in records:
|
|
_validate_record(record)
|
|
participation = record.resource_type == "ticket_participation"
|
|
actions.append(
|
|
DsarErasureActionRef(
|
|
action_id=f"tickets:{'manual_review' if participation else 'retain'}:{record.resource_type}:{record.resource_id}",
|
|
provider_id=self.provider_id,
|
|
module_id=self.module_id,
|
|
kind="manual_review" if participation else "retain",
|
|
resource_type=record.resource_type,
|
|
resource_id=record.resource_id,
|
|
title=f"Review {record.title}" if participation else f"Retain {record.title}",
|
|
rationale=(
|
|
"Reporter, requester, participant, and comment references may be operational evidence. "
|
|
"The ticket owner and applicable retention policy must decide whether they can be detached or minimized."
|
|
if participation
|
|
else record.retention_reason or "Ticket lifecycle attribution is immutable accountability evidence."
|
|
),
|
|
executable=False,
|
|
)
|
|
)
|
|
return tuple(actions)
|
|
|
|
def execute_erasure(
|
|
self,
|
|
session: object,
|
|
*,
|
|
tenant_id: str,
|
|
subject: DsarSubjectRef,
|
|
actions: Sequence[DsarErasureActionRef],
|
|
request_id: str,
|
|
) -> Sequence[DsarExecutionResultRef]:
|
|
del tenant_id
|
|
_session(session)
|
|
if _selectors(subject) is None:
|
|
raise ValueError("Tickets DSAR subject selectors conflict.")
|
|
results = []
|
|
for action in actions:
|
|
_validate_action(action)
|
|
if action.executable or action.kind not in {"retain", "manual_review"}:
|
|
raise ValueError("Tickets DSAR publishes non-executable actions only.")
|
|
results.append(
|
|
DsarExecutionResultRef(
|
|
action_id=action.action_id,
|
|
status="blocked",
|
|
summary=(
|
|
"The ticket remains unchanged pending owner and retention review."
|
|
if action.kind == "manual_review"
|
|
else "Ticket lifecycle attribution remains immutable evidence."
|
|
),
|
|
evidence={"request_id": request_id},
|
|
)
|
|
)
|
|
return tuple(results)
|
|
|
|
|
|
def _selectors(subject: DsarSubjectRef) -> _Selectors | None:
|
|
refs = subject.external_references
|
|
values = (
|
|
_coalesce(subject.account_id, refs.get("tickets.account"), refs.get("access.account")),
|
|
_coalesce(subject.identity_id, refs.get("tickets.identity"), refs.get("identity.id")),
|
|
_coalesce(subject.membership_id, refs.get("tickets.membership"), refs.get("tenancy.membership")),
|
|
)
|
|
ticket = _coalesce(refs.get("tickets.ticket"), refs.get("tickets.item"))
|
|
if any(value is _CONFLICT for value in (*values, ticket)):
|
|
return None
|
|
subject_ids = tuple(dict.fromkeys(value for value in values if isinstance(value, str) and value))
|
|
if not subject_ids:
|
|
return None
|
|
return _Selectors(subject_ids=subject_ids, ticket_id=ticket if isinstance(ticket, str) else None)
|
|
|
|
|
|
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 _subject_roles(row: Ticket, subject_ids: Sequence[str]) -> list[dict[str, object]]:
|
|
selected = set(subject_ids)
|
|
roles: list[dict[str, object]] = []
|
|
for role, value in (
|
|
("reporter", row.reporter),
|
|
("requester", row.requester),
|
|
("assignee", row.assignee),
|
|
):
|
|
if isinstance(value, Mapping) and str(value.get("id") or "") in selected:
|
|
roles.append({"role": role, "kind": value.get("kind"), "id": value.get("id"), "label": value.get("label")})
|
|
for value in row.participants or ():
|
|
if isinstance(value, Mapping) and str(value.get("id") or "") in selected:
|
|
roles.append({"role": value.get("role") or "participant", "kind": value.get("kind"), "id": value.get("id"), "label": value.get("label")})
|
|
return roles[:100]
|
|
|
|
|
|
def _activities(session: Session, row: Ticket, subject_ids: Sequence[str]) -> list[dict[str, object]]:
|
|
selected = set(subject_ids)
|
|
history = (
|
|
session.query(TicketHistory)
|
|
.filter(
|
|
TicketHistory.tenant_id == row.tenant_id,
|
|
TicketHistory.ticket_id == row.id,
|
|
TicketHistory.actor_id.in_(tuple(selected)),
|
|
)
|
|
.order_by(TicketHistory.revision.asc())
|
|
.limit(500)
|
|
.all()
|
|
)
|
|
return [
|
|
{"event_type": item.event_type, "revision": item.revision, "occurred_at": _iso(item.occurred_at)}
|
|
for item in history
|
|
]
|
|
|
|
|
|
def _comments(session: Session, row: Ticket, subject_ids: Sequence[str]) -> list[dict[str, object]]:
|
|
comments = (
|
|
session.query(TicketComment)
|
|
.filter(
|
|
TicketComment.tenant_id == row.tenant_id,
|
|
TicketComment.ticket_id == row.id,
|
|
TicketComment.created_by.in_(tuple(subject_ids)),
|
|
)
|
|
.order_by(TicketComment.created_at.asc())
|
|
.limit(500)
|
|
.all()
|
|
)
|
|
return [
|
|
{
|
|
"comment_id": item.comment_id,
|
|
"body": item.body[:20_000],
|
|
"visibility": item.visibility,
|
|
"created_at": _iso(item.created_at),
|
|
}
|
|
for item in comments
|
|
]
|
|
|
|
|
|
def _participation_record(
|
|
row: Ticket,
|
|
roles: list[dict[str, object]],
|
|
comments: list[dict[str, object]],
|
|
activities: list[dict[str, object]],
|
|
) -> DsarRecordRef:
|
|
return DsarRecordRef(
|
|
provider_id="tickets",
|
|
module_id="tickets",
|
|
resource_type="ticket_participation",
|
|
resource_id=row.id,
|
|
category="operational_service_request",
|
|
title=f"Ticket participation: {row.ticket_number}",
|
|
data={
|
|
"ticket_number": row.ticket_number,
|
|
"ticket_type": row.ticket_type,
|
|
"priority": row.priority,
|
|
"status": row.status,
|
|
"title": row.title[:500],
|
|
"description": row.description[:40_000],
|
|
"subject_roles": roles,
|
|
"subject_comments": comments,
|
|
"subject_activities": activities,
|
|
"received_at": _iso(row.received_at),
|
|
"resolved_at": _iso(row.resolved_at),
|
|
"deleted_at": _iso(row.deleted_at),
|
|
"revision": row.revision,
|
|
},
|
|
observed_at=_aware(row.updated_at or row.recorded_at),
|
|
retention_reason="The ticket may document institutional service delivery and requires owner review before subject references are changed.",
|
|
)
|
|
|
|
|
|
def _actor_record(row: Ticket, activities: list[dict[str, object]]) -> DsarRecordRef:
|
|
return DsarRecordRef(
|
|
provider_id="tickets",
|
|
module_id="tickets",
|
|
resource_type="ticket_actor_attribution",
|
|
resource_id=row.id,
|
|
category="operator_accountability_evidence",
|
|
title="Ticket lifecycle attribution",
|
|
data={"activities": activities, "status": row.status, "revision": row.revision},
|
|
observed_at=_aware(row.updated_at or row.recorded_at),
|
|
immutable_evidence=True,
|
|
retention_reason="Ticket lifecycle attribution is immutable accountability evidence.",
|
|
)
|
|
|
|
|
|
def _aware(value: datetime | None) -> datetime | None:
|
|
if value is not None and value.tzinfo is None:
|
|
return value.replace(tzinfo=timezone.utc)
|
|
return value
|
|
|
|
|
|
def _iso(value: datetime | None) -> str | None:
|
|
result = _aware(value)
|
|
return result.isoformat() if result else None
|
|
|
|
|
|
def _session(value: object) -> Session:
|
|
if not isinstance(value, Session):
|
|
raise TypeError("Tickets DSAR requires a SQLAlchemy Session.")
|
|
return value
|
|
|
|
|
|
def _validate_record(record: DsarRecordRef) -> None:
|
|
if record.provider_id != "tickets" or record.module_id != "tickets" or record.resource_type not in {"ticket_participation", "ticket_actor_attribution"}:
|
|
raise ValueError("Tickets DSAR record identity is invalid.")
|
|
|
|
|
|
def _validate_action(action: DsarErasureActionRef) -> None:
|
|
if action.provider_id != "tickets" or action.module_id != "tickets" or not action.action_id.startswith("tickets:"):
|
|
raise ValueError("Tickets DSAR action identity is invalid.")
|
|
|
|
|
|
__all__ = ["TICKETS_DSAR_CAPABILITY", "TicketsDsarProvider"]
|