From 6e373dcdd61ae2b60bba296b39eee22c7270d383 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Mon, 20 Jul 2026 17:33:47 +0200 Subject: [PATCH] feat(poll): define the scheduling capability contract --- src/govoplan_core/core/poll.py | 215 +++++++++++++++++++++++++++++++++ 1 file changed, 215 insertions(+) create mode 100644 src/govoplan_core/core/poll.py diff --git a/src/govoplan_core/core/poll.py b/src/govoplan_core/core/poll.py new file mode 100644 index 0000000..9918dc4 --- /dev/null +++ b/src/govoplan_core/core/poll.py @@ -0,0 +1,215 @@ +from __future__ import annotations + +from collections.abc import Mapping, Sequence +from dataclasses import dataclass, field +from datetime import datetime +from typing import Protocol, runtime_checkable + + +CAPABILITY_POLL_SCHEDULING = "poll.scheduling" + + +class PollCapabilityError(ValueError): + """Stable error raised by Poll capability implementations.""" + + +@dataclass(frozen=True, slots=True) +class PollOptionRequest: + label: str + key: str | None = None + description: str | None = None + value: Mapping[str, object] | None = None + metadata: Mapping[str, object] = field(default_factory=dict) + + +@dataclass(frozen=True, slots=True) +class PollCreateCommand: + title: str + kind: str + status: str + visibility: str + result_visibility: str + description: str | None = None + context_module: str | None = None + context_resource_type: str | None = None + context_resource_id: str | None = None + workflow_state: str | None = None + workflow_steps: tuple[Mapping[str, object], ...] = () + allow_anonymous: bool = False + allow_response_update: bool = True + min_choices: int = 1 + max_choices: int | None = None + opens_at: datetime | None = None + closes_at: datetime | None = None + options: tuple[PollOptionRequest, ...] = () + metadata: Mapping[str, object] = field(default_factory=dict) + + +@dataclass(frozen=True, slots=True) +class PollUpdateCommand: + """Scheduling-owned Poll fields synchronized as one policy snapshot.""" + + title: str + description: str | None + visibility: str + result_visibility: str + allow_anonymous: bool + allow_response_update: bool + closes_at: datetime | None + + +@dataclass(frozen=True, slots=True) +class PollInvitationCommand: + respondent_id: str | None = None + respondent_label: str | None = None + email: str | None = None + expires_at: datetime | None = None + metadata: Mapping[str, object] = field(default_factory=dict) + + +@dataclass(frozen=True, slots=True) +class PollAnswerRequest: + option_id: str | None = None + option_key: str | None = None + value: object = None + rank: int | None = None + + +@dataclass(frozen=True, slots=True) +class PollSubmitResponseCommand: + respondent_id: str + respondent_label: str | None = None + answers: tuple[PollAnswerRequest, ...] = () + metadata: Mapping[str, object] = field(default_factory=dict) + + +@dataclass(frozen=True, slots=True) +class PollOptionRef: + id: str + position: int + + +@dataclass(frozen=True, slots=True) +class PollRef: + id: str + status: str + options: tuple[PollOptionRef, ...] = () + + +@dataclass(frozen=True, slots=True) +class PollInvitationRef: + id: str + token: str + + +@dataclass(frozen=True, slots=True) +class PollResponseRef: + invitation_id: str | None + submitted_at: datetime + # Optional for backwards compatibility with existing providers. A + # consumer can use it to reconcile an authenticated response without + # trusting client-supplied invitation metadata. + respondent_id: str | None = None + + +@runtime_checkable +class PollSchedulingProvider(Protocol): + def create_poll( + self, + session: object, + *, + tenant_id: str, + user_id: str | None, + command: PollCreateCommand, + ) -> PollRef: + ... + + def create_invitation( + self, + session: object, + *, + tenant_id: str, + poll_id: str, + command: PollInvitationCommand, + ) -> PollInvitationRef: + ... + + def update_poll( + self, + session: object, + *, + tenant_id: str, + poll_id: str, + command: PollUpdateCommand, + ) -> PollRef: + ... + + def open_poll(self, session: object, *, tenant_id: str, poll_id: str) -> PollRef: + ... + + def close_poll(self, session: object, *, tenant_id: str, poll_id: str) -> PollRef: + ... + + def decide_poll( + self, + session: object, + *, + tenant_id: str, + poll_id: str, + option_id: str | None, + ) -> PollRef: + ... + + def get_poll(self, session: object, *, tenant_id: str, poll_id: str) -> PollRef: + ... + + def set_workflow_context( + self, + session: object, + *, + tenant_id: str, + poll_id: str, + workflow_state: str, + workflow_steps: Sequence[Mapping[str, object]], + context_module: str, + context_resource_type: str, + context_resource_id: str, + ) -> PollRef: + ... + + def result_summary(self, session: object, *, tenant_id: str, poll_id: str) -> Mapping[str, object]: + ... + + def list_responses(self, session: object, *, tenant_id: str, poll_id: str) -> Sequence[PollResponseRef]: + ... + + +@runtime_checkable +class PollResponseSubmissionProvider(Protocol): + """Optional extension for modules that collect responses through Poll.""" + + def submit_response( + self, + session: object, + *, + tenant_id: str, + poll_id: str, + command: PollSubmitResponseCommand, + ) -> PollResponseRef: + ... + + +def poll_scheduling_provider(registry: object | None) -> PollSchedulingProvider | None: + if registry is None or not hasattr(registry, "has_capability"): + return None + if not registry.has_capability(CAPABILITY_POLL_SCHEDULING): + return None + capability = registry.capability(CAPABILITY_POLL_SCHEDULING) + return capability if isinstance(capability, PollSchedulingProvider) else None + + +def poll_response_submission_provider( + registry: object | None, +) -> PollResponseSubmissionProvider | None: + provider = poll_scheduling_provider(registry) + return provider if isinstance(provider, PollResponseSubmissionProvider) else None