from __future__ import annotations from dataclasses import dataclass from types import MappingProxyType from typing import Mapping POLL_STATUSES = frozenset({"draft", "open", "closed", "decided", "archived"}) POLL_TRANSITION_ACTIONS = ("open", "draft", "close", "decide", "archive", "unarchive") @dataclass(frozen=True) class PollTransitionRule: """One configurable lifecycle action. Archive restoration is deliberately represented as a dynamic target. The engine remains a pure policy component; persistence and domain side effects stay in the Poll service. """ action: str source_statuses: frozenset[str] target_status: str | None restore_archived_status: bool = False missing_restore_target_status: str | None = None @dataclass(frozen=True) class PollTransitionPlan: action: str from_status: str to_status: str @dataclass(frozen=True) class PollTransitionAvailability: action: str target_status: str | None available: bool reason: str | None = None class PollTransitionPolicy: """Validated, replaceable configuration for a Poll lifecycle.""" def __init__(self, rules: Mapping[str, PollTransitionRule]) -> None: normalized = dict(rules) if set(normalized) != set(POLL_TRANSITION_ACTIONS): missing = sorted(set(POLL_TRANSITION_ACTIONS) - set(normalized)) extra = sorted(set(normalized) - set(POLL_TRANSITION_ACTIONS)) raise ValueError(f"Poll transition policy has missing actions {missing} and extra actions {extra}") for action, rule in normalized.items(): if rule.action != action: raise ValueError(f"Poll transition rule key {action!r} does not match action {rule.action!r}") if not rule.source_statuses <= POLL_STATUSES: raise ValueError(f"Poll transition rule {action!r} has unknown source statuses") if rule.restore_archived_status: if rule.target_status is not None: raise ValueError(f"Restoring transition {action!r} cannot have a fixed target") if rule.missing_restore_target_status not in (POLL_STATUSES - {"archived"}) | {None}: raise ValueError(f"Restoring transition {action!r} has an invalid missing-origin target") elif rule.target_status not in POLL_STATUSES: raise ValueError(f"Poll transition rule {action!r} has an unknown target status") elif rule.missing_restore_target_status is not None: raise ValueError(f"Non-restoring transition {action!r} cannot have a missing-origin target") self._rules = MappingProxyType(normalized) @property def rules(self) -> Mapping[str, PollTransitionRule]: return self._rules DEFAULT_POLL_TRANSITION_POLICY = PollTransitionPolicy( { "open": PollTransitionRule( action="open", source_statuses=frozenset({"draft", "closed", "decided"}), target_status="open", ), "draft": PollTransitionRule( action="draft", source_statuses=frozenset({"open"}), target_status="draft", ), "close": PollTransitionRule( action="close", source_statuses=frozenset({"open"}), target_status="closed", ), "decide": PollTransitionRule( action="decide", source_statuses=frozenset({"closed", "decided"}), target_status="decided", ), "archive": PollTransitionRule( action="archive", source_statuses=frozenset({"draft", "open", "closed", "decided"}), target_status="archived", ), "unarchive": PollTransitionRule( action="unarchive", source_statuses=frozenset({"archived"}), target_status=None, restore_archived_status=True, missing_restore_target_status="draft", ), } ) class PollTransitionEngine: """Pure lifecycle planner backed by an injected transition policy.""" def __init__(self, policy: PollTransitionPolicy = DEFAULT_POLL_TRANSITION_POLICY) -> None: self.policy = policy def plan( self, *, current_status: str, action: str, archived_from_status: str | None = None, ) -> PollTransitionPlan: if current_status not in POLL_STATUSES: raise ValueError(f"Unknown poll status: {current_status}") rule = self.policy.rules.get(action) if rule is None: raise ValueError(f"Unknown poll transition action: {action}") if current_status not in rule.source_statuses: raise ValueError(f"Poll transition {action!r} is not allowed from status {current_status!r}") target_status = rule.target_status if rule.restore_archived_status: if archived_from_status is None and rule.missing_restore_target_status is not None: target_status = rule.missing_restore_target_status elif archived_from_status not in POLL_STATUSES - {"archived"}: raise ValueError("Archived poll has no valid status to restore") else: target_status = archived_from_status if target_status is None: raise ValueError(f"Poll transition {action!r} did not resolve a target status") return PollTransitionPlan(action=action, from_status=current_status, to_status=target_status) def available_actions( self, *, current_status: str, archived_from_status: str | None = None, ) -> tuple[PollTransitionAvailability, ...]: actions: list[PollTransitionAvailability] = [] for action in POLL_TRANSITION_ACTIONS: rule = self.policy.rules[action] target_status = ( archived_from_status or rule.missing_restore_target_status if rule.restore_archived_status else rule.target_status ) try: self.plan( current_status=current_status, action=action, archived_from_status=archived_from_status, ) except ValueError as exc: actions.append( PollTransitionAvailability( action=action, target_status=target_status, available=False, reason=str(exc), ) ) else: actions.append( PollTransitionAvailability( action=action, target_status=target_status, available=True, ) ) return tuple(actions) DEFAULT_POLL_TRANSITION_ENGINE = PollTransitionEngine() __all__ = [ "DEFAULT_POLL_TRANSITION_ENGINE", "DEFAULT_POLL_TRANSITION_POLICY", "POLL_STATUSES", "POLL_TRANSITION_ACTIONS", "PollTransitionAvailability", "PollTransitionEngine", "PollTransitionPlan", "PollTransitionPolicy", "PollTransitionRule", ]