GovOPlaN/govoplan-poll
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(poll): implement core participation contract

Browse Source
This commit is contained in:
Albrecht Degering
2026-07-22 03:21:08 +02:00
parent eb003208e4
commit 38ecff60a5
4 changed files with 58 additions and 251 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
pyproject.toml
View File

@@ -11,7 +11,7 @@ requires-python = ">=3.12"
license = { file = "LICENSE" }
authors = [{ name = "GovOPlaN" }]
dependencies = [
"govoplan-core>=0.1.9",
"govoplan-core>=0.1.10",
]
[tool.setuptools.packages.find]

4
src/govoplan_poll/backend/manifest.py
View File

@@ -13,9 +13,9 @@ from govoplan_core.core.modules import (
PermissionDefinition,
RoleTemplate,
)
from govoplan_core.db.base import Base
from govoplan_core.core.poll import CAPABILITY_POLL_SCHEDULING
from govoplan_poll.backend.participation import CAPABILITY_POLL_PARTICIPATION_GATEWAY
from govoplan_core.core.poll_participation import CAPABILITY_POLL_PARTICIPATION_GATEWAY
from govoplan_core.db.base import Base
from govoplan_poll.backend.db import models as poll_models # noqa: F401 - populate Poll ORM metadata
MODULE_ID = "poll"

271
src/govoplan_poll/backend/participation.py
View File

@@ -1,254 +1,29 @@
from __future__ import annotations
"""Backward-compatible governed-participation contract imports.
import hashlib
from collections.abc import Mapping
from dataclasses import dataclass, field
from datetime import datetime
from typing import Protocol, runtime_checkable
The provider contract belongs to Core so consumers can resolve Poll through
the platform capability registry without importing Poll implementation
internals. This module intentionally preserves the original public import
path for existing Poll integrations.
"""
from govoplan_core.core.poll import (
PollAnswerRequest,
PollInvitationRef,
PollOptionRequest,
PollResponseRef,
from govoplan_core.core.poll_participation import (
ANONYMOUS_PASSWORD_REQUIREMENT,
CAPABILITY_POLL_PARTICIPATION_GATEWAY,
PARTICIPATION_POLICY_VERSION,
PollGovernedInvitationCommand,
PollGovernedResponseCommand,
PollGovernedResponseRef,
PollInvitationExpiryRef,
PollInvitationRevocationRef,
PollOptionMutationRef,
PollParticipationContextRef,
PollParticipationGatewayProvider,
PollParticipationPolicy,
PollResponseGatewayRef,
participation_token_fingerprint,
poll_participation_gateway_provider,
)
CAPABILITY_POLL_PARTICIPATION_GATEWAY = "poll.participation_gateway"
# Policy-attestation identifier, not a credential or credential default.
ANONYMOUS_PASSWORD_REQUIREMENT = "anonymous_password" # nosec B105 # noqa: S105
PARTICIPATION_POLICY_VERSION = 1
def participation_token_fingerprint(token: str) -> str:
"""Return a non-reversible identifier suitable for audit/throttle keys."""
return hashlib.sha256(token.encode("utf-8")).hexdigest()
@dataclass(frozen=True, slots=True)
class PollResponseGatewayRef:
"""Stable identity of the module resource governing a participation link."""
module_id: str
resource_type: str
resource_id: str
@dataclass(frozen=True, slots=True)
class PollParticipationPolicy:
"""Generic response rules snapshotted onto one signed invitation.
``single_choice`` treats every non-``unavailable`` availability answer as
a selection. Capacity is reserved only by ``available`` answers (and by
selected answers for non-availability polls), so ``maybe`` never consumes
a place.
"""
version: int = PARTICIPATION_POLICY_VERSION
single_choice: bool = False
allow_maybe: bool = True
max_participants_per_option: int | None = None
allow_comments: bool = False
participant_email_required: bool = False
anonymous_password_required: bool = False
@dataclass(frozen=True, slots=True)
class PollGovernedInvitationCommand:
gateway: PollResponseGatewayRef
policy: PollParticipationPolicy
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 PollGovernedResponseCommand:
"""Submission already authorized by the named in-process gateway.
Passwords never cross this boundary. A gateway that owns a password
verifier reports the completed check through ``verified_requirements``.
Poll independently re-enforces the remaining snapshotted rules while
holding its Poll-row lock.
"""
respondent_id: str | None = None
respondent_label: str | None = None
participant_email: str | None = None
participant_is_authenticated: bool = False
answers: tuple[PollAnswerRequest, ...] = ()
comment: str | None = None
verified_requirements: frozenset[str] = frozenset()
idempotency_key: str | None = None
metadata: Mapping[str, object] = field(default_factory=dict)
@dataclass(frozen=True, slots=True)
class PollGovernedResponseRef:
response: PollResponseRef
participant_email: str | None = None
comment: str | None = None
replayed: bool = False
@dataclass(frozen=True, slots=True)
class PollOptionMutationRef:
id: str
position: int
replayed: bool = False
invalidated_response_count: int = 0
@dataclass(frozen=True, slots=True)
class PollInvitationRevocationRef:
id: str
revoked_at: datetime
replayed: bool = False
@dataclass(frozen=True, slots=True)
class PollInvitationExpiryRef:
id: str
expires_at: datetime | None
replayed: bool = False
@dataclass(frozen=True, slots=True)
class PollParticipationContextRef:
invitation_id: str
tenant_id: str
poll_id: str
gateway: PollResponseGatewayRef
policy: PollParticipationPolicy
respondent_id: str | None = None
respondent_label: str | None = None
email: str | None = None
response: PollGovernedResponseRef | None = None
@runtime_checkable
class PollParticipationGatewayProvider(Protocol):
def create_governed_invitation(
self,
session: object,
*,
tenant_id: str,
poll_id: str,
command: PollGovernedInvitationCommand,
) -> PollInvitationRef:
...
def resolve_participation(
self,
session: object,
*,
token: str,
gateway: PollResponseGatewayRef,
respondent_id: str | None = None,
participant_email: str | None = None,
participant_is_authenticated: bool = False,
verified_requirements: frozenset[str] = frozenset(),
) -> PollParticipationContextRef:
"""Resolve and prefill one valid invitation for the exact gateway."""
...
def submit_governed_response(
self,
session: object,
*,
token: str,
gateway: PollResponseGatewayRef,
command: PollGovernedResponseCommand,
) -> PollGovernedResponseRef:
...
def resolve_authenticated_participation(
self,
session: object,
*,
tenant_id: str,
poll_id: str,
invitation_id: str,
gateway: PollResponseGatewayRef,
respondent_id: str,
) -> PollParticipationContextRef:
"""Resolve one governed invitation without retaining its public token."""
...
def submit_authenticated_response(
self,
session: object,
*,
tenant_id: str,
poll_id: str,
invitation_id: str,
gateway: PollResponseGatewayRef,
respondent_id: str,
command: PollGovernedResponseCommand,
) -> PollGovernedResponseRef:
"""Submit atomically for the exact authenticated invitation identity."""
...
def add_option(
self,
session: object,
*,
tenant_id: str,
poll_id: str,
command: PollOptionRequest,
) -> PollOptionMutationRef:
...
def remove_option(
self,
session: object,
*,
tenant_id: str,
poll_id: str,
option_id: str,
) -> PollOptionMutationRef:
...
def revoke_invitation(
self,
session: object,
*,
tenant_id: str,
poll_id: str,
invitation_id: str,
) -> PollInvitationRevocationRef:
...
def update_invitation_expiry(
self,
session: object,
*,
tenant_id: str,
poll_id: str,
invitation_id: str,
gateway: PollResponseGatewayRef,
expires_at: datetime | None,
) -> PollInvitationExpiryRef:
"""Expire or extend a non-revoked governed invitation in place."""
...
def poll_participation_gateway_provider(registry: object | None) -> PollParticipationGatewayProvider | None:
if registry is None or not hasattr(registry, "has_capability"):
return None
if not registry.has_capability(CAPABILITY_POLL_PARTICIPATION_GATEWAY):
return None
capability = registry.capability(CAPABILITY_POLL_PARTICIPATION_GATEWAY)
return capability if isinstance(capability, PollParticipationGatewayProvider) else None
__all__ = [
"ANONYMOUS_PASSWORD_REQUIREMENT",
"CAPABILITY_POLL_PARTICIPATION_GATEWAY",
@@ -256,8 +31,8 @@ __all__ = [
"PollGovernedInvitationCommand",
"PollGovernedResponseCommand",
"PollGovernedResponseRef",
"PollInvitationRevocationRef",
"PollInvitationExpiryRef",
"PollInvitationRevocationRef",
"PollOptionMutationRef",
"PollParticipationContextRef",
"PollParticipationGatewayProvider",

32
tests/test_participation_contract.py Normal file
View File

@@ -0,0 +1,32 @@
from __future__ import annotations
import unittest
from govoplan_core.core.poll_participation import (
CAPABILITY_POLL_PARTICIPATION_GATEWAY as CORE_CAPABILITY,
PollGovernedInvitationCommand as CorePollGovernedInvitationCommand,
PollParticipationGatewayProvider as CorePollParticipationGatewayProvider,
PollParticipationPolicy as CorePollParticipationPolicy,
)
from govoplan_poll.backend.capabilities import SqlPollSchedulingProvider
from govoplan_poll.backend.participation import (
CAPABILITY_POLL_PARTICIPATION_GATEWAY,
PollGovernedInvitationCommand,
PollParticipationGatewayProvider,
PollParticipationPolicy,
)
class PollParticipationCompatibilityTests(unittest.TestCase):
def test_legacy_imports_are_exact_core_contract_re_exports(self) -> None:
self.assertIs(CORE_CAPABILITY, CAPABILITY_POLL_PARTICIPATION_GATEWAY)
self.assertIs(CorePollGovernedInvitationCommand, PollGovernedInvitationCommand)
self.assertIs(CorePollParticipationPolicy, PollParticipationPolicy)
self.assertIs(CorePollParticipationGatewayProvider, PollParticipationGatewayProvider)
def test_sql_provider_implements_the_core_contract(self) -> None:
self.assertIsInstance(SqlPollSchedulingProvider(), CorePollParticipationGatewayProvider)
if __name__ == "__main__":
unittest.main()