372 lines
11 KiB
Python
372 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
from collections.abc import Mapping, Sequence
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
from typing import Any, Literal, Protocol
|
|
|
|
|
|
MAX_RETIREMENT_RESPONDENT_IDS = 500
|
|
MAX_RETIREMENT_RESPONSES = 1000
|
|
OWNERSHIP_FIELDS = frozenset(
|
|
{
|
|
"context_module",
|
|
"context_resource_type",
|
|
"context_resource_id",
|
|
"workflow_state",
|
|
"workflow_steps",
|
|
}
|
|
)
|
|
DIRECT_UPDATE_FIELDS = (
|
|
"title",
|
|
"description",
|
|
"visibility",
|
|
"result_visibility",
|
|
"context_module",
|
|
"context_resource_type",
|
|
"context_resource_id",
|
|
"workflow_state",
|
|
"allow_anonymous",
|
|
"allow_response_update",
|
|
)
|
|
|
|
ResponseDisposition = Literal[
|
|
"preserve",
|
|
"invalidate_affected_answers",
|
|
"retire",
|
|
"reject",
|
|
]
|
|
|
|
|
|
class PollMutationPlanError(ValueError):
|
|
pass
|
|
|
|
|
|
class PollLike(Protocol):
|
|
status: str
|
|
kind: str
|
|
min_choices: int
|
|
max_choices: int | None
|
|
opens_at: datetime | None
|
|
closes_at: datetime | None
|
|
|
|
|
|
class RetirableResponse(Protocol):
|
|
id: str
|
|
deleted_at: datetime | None
|
|
metadata_: dict[str, Any] | None
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ExistingResponseDecision:
|
|
change: str
|
|
disposition: ResponseDisposition
|
|
reason: str
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class PollUpdatePlan:
|
|
values: Mapping[str, object]
|
|
response_decision: ExistingResponseDecision
|
|
|
|
def apply(self, poll: object) -> None:
|
|
for field, value in self.values.items():
|
|
setattr(poll, field, value)
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ResponseRetirementSelector:
|
|
respondent_ids: tuple[str, ...]
|
|
invitation_id: str | None
|
|
reason: str
|
|
idempotency_key: str
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ResponseRetirementPlan:
|
|
responses: tuple[RetirableResponse, ...]
|
|
retired_at: datetime | None
|
|
disposition: Literal["retire", "replay", "noop"]
|
|
|
|
@property
|
|
def newly_retired_count(self) -> int:
|
|
return len(self.responses) if self.disposition == "retire" else 0
|
|
|
|
def apply(
|
|
self,
|
|
*,
|
|
reason: str,
|
|
idempotency_key: str,
|
|
metadata: Mapping[str, object],
|
|
) -> None:
|
|
if self.disposition != "retire" or self.retired_at is None:
|
|
return
|
|
retirement = {
|
|
"idempotency_key": idempotency_key,
|
|
"reason": reason,
|
|
"retired_at": self.retired_at.isoformat(),
|
|
"context": dict(metadata),
|
|
}
|
|
for response in self.responses:
|
|
response.metadata_ = {
|
|
**(response.metadata_ or {}),
|
|
"response_retirement": retirement,
|
|
}
|
|
response.deleted_at = self.retired_at
|
|
|
|
|
|
def plan_poll_update(
|
|
poll: PollLike,
|
|
values: Mapping[str, object],
|
|
*,
|
|
active_option_count: int,
|
|
) -> PollUpdatePlan:
|
|
if poll.status in {"closed", "decided", "archived"}:
|
|
raise PollMutationPlanError(
|
|
"Closed, decided, or archived polls cannot be edited"
|
|
)
|
|
|
|
updates: dict[str, object] = {}
|
|
for field in DIRECT_UPDATE_FIELDS:
|
|
value = values.get(field)
|
|
if value is not None:
|
|
updates[field] = value
|
|
for field in ("workflow_steps", "metadata"):
|
|
value = values.get(field)
|
|
if value is not None:
|
|
updates["metadata_" if field == "metadata" else field] = value
|
|
|
|
min_choices = (
|
|
poll.min_choices
|
|
if values.get("min_choices") is None
|
|
else int(values["min_choices"]) # type: ignore[arg-type]
|
|
)
|
|
max_choices = (
|
|
poll.max_choices
|
|
if values.get("max_choices") is None
|
|
else int(values["max_choices"]) # type: ignore[arg-type]
|
|
)
|
|
if values.get("min_choices") is not None or values.get("max_choices") is not None:
|
|
min_choices, max_choices = validate_choice_bounds(
|
|
poll.kind,
|
|
min_choices,
|
|
max_choices,
|
|
active_option_count,
|
|
)
|
|
updates["min_choices"] = min_choices
|
|
updates["max_choices"] = max_choices
|
|
|
|
opens_at = (
|
|
values["opens_at"]
|
|
if values.get("opens_at") is not None
|
|
else poll.opens_at
|
|
)
|
|
closes_at = (
|
|
values["closes_at"]
|
|
if values.get("closes_at") is not None
|
|
else poll.closes_at
|
|
)
|
|
if values.get("opens_at") is not None:
|
|
updates["opens_at"] = opens_at
|
|
if values.get("closes_at") is not None:
|
|
updates["closes_at"] = closes_at
|
|
if (
|
|
isinstance(opens_at, datetime)
|
|
and isinstance(closes_at, datetime)
|
|
and _comparable_datetime(closes_at) <= _comparable_datetime(opens_at)
|
|
):
|
|
raise PollMutationPlanError("closes_at must be after opens_at")
|
|
|
|
return PollUpdatePlan(
|
|
values=updates,
|
|
response_decision=ExistingResponseDecision(
|
|
change="poll_policy_or_scope",
|
|
disposition="preserve",
|
|
reason=(
|
|
"Poll metadata, policy, timing, and owner-approved scope changes "
|
|
"do not alter stable option identities or submitted answers."
|
|
),
|
|
),
|
|
)
|
|
|
|
|
|
def decide_existing_response_impact(
|
|
change: Literal[
|
|
"option_content",
|
|
"option_remove",
|
|
"option_reorder",
|
|
"participant_remove",
|
|
"poll_policy_or_scope",
|
|
],
|
|
*,
|
|
has_active_responses: bool,
|
|
allow_response_update: bool,
|
|
) -> ExistingResponseDecision:
|
|
if not has_active_responses:
|
|
return ExistingResponseDecision(
|
|
change=change,
|
|
disposition="preserve",
|
|
reason="No active responses are affected.",
|
|
)
|
|
if change in {"option_content", "option_remove"}:
|
|
if not allow_response_update:
|
|
return ExistingResponseDecision(
|
|
change=change,
|
|
disposition="reject",
|
|
reason=(
|
|
"Poll options cannot be edited after responses when "
|
|
"response updates are disabled"
|
|
),
|
|
)
|
|
return ExistingResponseDecision(
|
|
change=change,
|
|
disposition="invalidate_affected_answers",
|
|
reason=(
|
|
"Only answers bound to the changed stable option identity are "
|
|
"invalidated; empty responses are retired."
|
|
),
|
|
)
|
|
if change == "participant_remove":
|
|
return ExistingResponseDecision(
|
|
change=change,
|
|
disposition="retire",
|
|
reason="Responses for the removed participant leave live results.",
|
|
)
|
|
return ExistingResponseDecision(
|
|
change=change,
|
|
disposition="preserve",
|
|
reason="Stable response and option identities remain valid.",
|
|
)
|
|
|
|
|
|
def normalize_retirement_selector(
|
|
*,
|
|
respondent_ids: Sequence[str],
|
|
invitation_id: str | None,
|
|
reason: str,
|
|
idempotency_key: str,
|
|
) -> ResponseRetirementSelector:
|
|
normalized_ids = tuple(
|
|
dict.fromkeys(value.strip() for value in respondent_ids if value.strip())
|
|
)
|
|
if len(normalized_ids) > MAX_RETIREMENT_RESPONDENT_IDS:
|
|
raise PollMutationPlanError(
|
|
"Response retirement targets too many participant identities"
|
|
)
|
|
normalized_invitation_id = (
|
|
invitation_id.strip()
|
|
if invitation_id is not None and invitation_id.strip()
|
|
else None
|
|
)
|
|
normalized_reason = reason.strip()
|
|
normalized_key = idempotency_key.strip()
|
|
if not normalized_ids and normalized_invitation_id is None:
|
|
raise PollMutationPlanError(
|
|
"Response retirement requires a trusted participant identity"
|
|
)
|
|
if not normalized_reason or len(normalized_reason) > 120:
|
|
raise PollMutationPlanError("Response retirement reason is invalid")
|
|
if not normalized_key or len(normalized_key) > 255:
|
|
raise PollMutationPlanError(
|
|
"Response retirement idempotency key is invalid"
|
|
)
|
|
return ResponseRetirementSelector(
|
|
respondent_ids=normalized_ids,
|
|
invitation_id=normalized_invitation_id,
|
|
reason=normalized_reason,
|
|
idempotency_key=normalized_key,
|
|
)
|
|
|
|
|
|
def plan_response_retirement(
|
|
responses: Sequence[RetirableResponse],
|
|
*,
|
|
idempotency_key: str,
|
|
now: datetime,
|
|
) -> ResponseRetirementPlan:
|
|
replayed = tuple(
|
|
response
|
|
for response in responses
|
|
if isinstance((response.metadata_ or {}).get("response_retirement"), dict)
|
|
and (response.metadata_ or {})["response_retirement"].get(
|
|
"idempotency_key"
|
|
)
|
|
== idempotency_key
|
|
)
|
|
if replayed:
|
|
retired_at = max(
|
|
(
|
|
_comparable_datetime(response.deleted_at)
|
|
for response in replayed
|
|
if response.deleted_at is not None
|
|
),
|
|
default=None,
|
|
)
|
|
return ResponseRetirementPlan(
|
|
responses=replayed,
|
|
retired_at=retired_at,
|
|
disposition="replay",
|
|
)
|
|
active = tuple(response for response in responses if response.deleted_at is None)
|
|
if active:
|
|
return ResponseRetirementPlan(
|
|
responses=active,
|
|
retired_at=now,
|
|
disposition="retire",
|
|
)
|
|
return ResponseRetirementPlan(
|
|
responses=(),
|
|
retired_at=None,
|
|
disposition="noop",
|
|
)
|
|
|
|
|
|
def validate_choice_bounds(
|
|
kind: str,
|
|
min_choices: int,
|
|
max_choices: int | None,
|
|
option_count: int,
|
|
) -> tuple[int, int | None]:
|
|
if kind in {"single_choice", "yes_no", "yes_no_maybe"}:
|
|
return 1, 1
|
|
if kind == "ranked_choice":
|
|
min_choices = max(1, min_choices)
|
|
if max_choices is None:
|
|
max_choices = option_count
|
|
if min_choices > option_count:
|
|
raise PollMutationPlanError(
|
|
"min_choices cannot be greater than the number of options"
|
|
)
|
|
if max_choices is not None:
|
|
if max_choices < min_choices:
|
|
raise PollMutationPlanError(
|
|
"max_choices cannot be smaller than min_choices"
|
|
)
|
|
if max_choices > option_count:
|
|
raise PollMutationPlanError(
|
|
"max_choices cannot be greater than the number of options"
|
|
)
|
|
return min_choices, max_choices
|
|
|
|
|
|
def _comparable_datetime(value: datetime) -> datetime:
|
|
if value.tzinfo is None:
|
|
return value.replace(tzinfo=timezone.utc)
|
|
return value.astimezone(timezone.utc)
|
|
|
|
|
|
__all__ = [
|
|
"MAX_RETIREMENT_RESPONDENT_IDS",
|
|
"MAX_RETIREMENT_RESPONSES",
|
|
"ExistingResponseDecision",
|
|
"PollMutationPlanError",
|
|
"PollUpdatePlan",
|
|
"ResponseRetirementPlan",
|
|
"ResponseRetirementSelector",
|
|
"decide_existing_response_impact",
|
|
"normalize_retirement_selector",
|
|
"plan_poll_update",
|
|
"plan_response_retirement",
|
|
"validate_choice_bounds",
|
|
]
|