Files
govoplan-poll/src/govoplan_poll/backend/manifest.py
T

217 lines
8.9 KiB
Python

from __future__ import annotations
from pathlib import Path
from govoplan_core.core.access import CAPABILITY_AUTH_PERMISSION_EVALUATOR, CAPABILITY_AUTH_PRINCIPAL_RESOLVER
from govoplan_core.core.module_guards import drop_table_retirement_provider, persistent_table_uninstall_guard
from govoplan_core.core.modules import (
DocumentationTopic,
MigrationSpec,
ModuleContext,
ModuleInterfaceProvider,
ModuleManifest,
PermissionDefinition,
RoleTemplate,
)
from govoplan_core.core.provider_governance import declared_module_architecture
from govoplan_core.core.poll import CAPABILITY_POLL_SCHEDULING
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"
MODULE_NAME = "Poll"
MODULE_VERSION = "0.1.11"
READ_SCOPE = "poll:poll:read"
WRITE_SCOPE = "poll:poll:write"
ADMIN_SCOPE = "poll:poll:admin"
RESPOND_SCOPE = "poll:response:write"
def _permission(scope: str, label: str, description: str) -> PermissionDefinition:
module_id, resource, action = scope.split(":", 2)
return PermissionDefinition(
scope=scope,
label=label,
description=description,
category="Poll",
level="tenant",
module_id=module_id,
resource=resource,
action=action,
)
PERMISSIONS = (
_permission(READ_SCOPE, "View polls", "Read poll definitions, invitations, responses, and result summaries."),
_permission(WRITE_SCOPE, "Manage polls", "Create, edit, close, reopen, and decide polls."),
_permission(ADMIN_SCOPE, "Administer polls", "Configure tenant-level polling policies and visibility defaults."),
_permission(RESPOND_SCOPE, "Respond to polls", "Submit and update own poll responses where policy allows it."),
)
ROLE_TEMPLATES = (
RoleTemplate(
slug="poll_manager",
name="Poll manager",
description="Create, manage, close, and decide polls.",
permissions=(READ_SCOPE, WRITE_SCOPE, RESPOND_SCOPE),
),
RoleTemplate(
slug="poll_participant",
name="Poll participant",
description="Read assigned polls and submit own responses.",
permissions=(READ_SCOPE, RESPOND_SCOPE),
),
)
DOCUMENTATION = (
DocumentationTopic(
id="poll.module-boundary",
title="Poll module boundary",
summary="Lightweight decision and availability polls for reusable module integrations.",
body=(
"Poll owns reusable poll definitions, options, invitations, responses, visibility rules, "
"closing semantics, and result summaries. Scheduling consumes Poll for availability "
"matrices, while Evaluation owns heavier surveys, scoring, rubrics, and analytics. "
"Access is optional: when installed, Poll can use principal resolution, permission "
"evaluation, and role templates; without it, Poll is limited to anonymous, signed-link, "
"or adapter-provided participant flows."
),
layer="available",
documentation_types=("admin", "user"),
audience=("user", "operator", "module_admin", "product_owner"),
related_modules=("scheduling", "evaluation", "calendar", "campaigns", "portal"),
metadata={"seed": True},
),
DocumentationTopic(
id="poll.participation-and-results",
title="Respond to a poll",
summary="Polls can collect single or multiple choices, yes/no, yes/no/maybe, ranked choices, and availability responses.",
body=(
"An invitation or signed participation link determines which poll and participant identity a response belongs to. "
"The poll policy controls anonymity, response updates, result visibility, open and close times, and whether Maybe is allowed. "
"Submitting a response is atomic: capacity and choice constraints are checked before the saved response replaces any earlier answer. "
"A valid signed link also resolves its tenant before Poll runs, so tenant module policy can withdraw the public surface without exposing another tenant's state."
),
layer="configured",
documentation_types=("user",),
audience=("user", "participant", "organizer"),
related_modules=("scheduling", "evaluation"),
metadata={"kind": "reference"},
),
)
def _tenant_summary(session, tenant_id: str) -> dict[str, int]:
from govoplan_poll.backend.db.models import Poll, PollInvitation, PollResponse
return {
"polls": session.query(Poll).filter(Poll.tenant_id == tenant_id, Poll.deleted_at.is_(None)).count(),
"poll_invitations": session.query(PollInvitation).filter(PollInvitation.tenant_id == tenant_id).count(),
"poll_responses": session.query(PollResponse).filter(PollResponse.tenant_id == tenant_id, PollResponse.deleted_at.is_(None)).count(),
}
def _poll_router(_context: ModuleContext):
from govoplan_poll.backend.router import router
return router
def _public_tenant_resolver(request: object, session: object) -> str | None:
path_params = getattr(request, "path_params", {})
token = str(path_params.get("token") or "").strip()
path = str(getattr(getattr(request, "url", None), "path", ""))
if not token or "/poll/public/" not in path:
return None
from govoplan_poll.backend.service import PollError, get_poll_by_invitation_token
try:
poll = get_poll_by_invitation_token(session, token=token)
except PollError:
return None
return poll.tenant_id
def _poll_scheduling_provider(context: ModuleContext) -> object:
del context
from govoplan_poll.backend.capabilities import SqlPollSchedulingProvider
return SqlPollSchedulingProvider()
def _poll_participation_gateway_provider(context: ModuleContext) -> object:
return _poll_scheduling_provider(context)
manifest = ModuleManifest(
id=MODULE_ID,
name=MODULE_NAME,
version=MODULE_VERSION,
dependencies=(),
optional_dependencies=("access", "calendar", "campaigns", "portal", "mail", "notifications", "forms_runtime"),
optional_capabilities=(CAPABILITY_AUTH_PRINCIPAL_RESOLVER, CAPABILITY_AUTH_PERMISSION_EVALUATOR),
provides_interfaces=(
ModuleInterfaceProvider(name="poll.option_selection", version=MODULE_VERSION),
ModuleInterfaceProvider(name="poll.option_ordering", version=MODULE_VERSION),
ModuleInterfaceProvider(name="poll.availability_matrix", version=MODULE_VERSION),
ModuleInterfaceProvider(name="poll.response_collection", version=MODULE_VERSION),
ModuleInterfaceProvider(name="poll.workflow_context", version=MODULE_VERSION),
ModuleInterfaceProvider(name="poll.signed_participation", version=MODULE_VERSION),
ModuleInterfaceProvider(name="poll.governed_participation", version=MODULE_VERSION),
),
permissions=PERMISSIONS,
role_templates=ROLE_TEMPLATES,
route_factory=_poll_router,
public_tenant_resolver=_public_tenant_resolver,
tenant_summary_providers=(_tenant_summary,),
capability_factories={
CAPABILITY_POLL_SCHEDULING: _poll_scheduling_provider,
CAPABILITY_POLL_PARTICIPATION_GATEWAY: _poll_participation_gateway_provider,
},
migration_spec=MigrationSpec(
module_id=MODULE_ID,
metadata=Base.metadata,
script_location=str(Path(__file__).with_name("migrations") / "versions"),
retirement_supported=True,
retirement_provider=drop_table_retirement_provider(
poll_models.Poll,
poll_models.PollInvitation,
poll_models.PollLifecycleTransition,
poll_models.PollOption,
poll_models.PollParticipationSubmission,
poll_models.PollResponse,
label="Poll",
),
retirement_notes="Destructive retirement drops poll-owned database tables after the installer captures a database snapshot.",
),
uninstall_guard_providers=(
persistent_table_uninstall_guard(
poll_models.Poll,
poll_models.PollInvitation,
poll_models.PollLifecycleTransition,
poll_models.PollOption,
poll_models.PollParticipationSubmission,
poll_models.PollResponse,
label="Poll",
),
),
documentation=DOCUMENTATION,
architecture=declared_module_architecture(
layer="communication_participation",
kind="domain",
maturity="vertical_slice",
documentation_ref="README.md",
test_ref="tests/test_service.py",
known_limits=("Advanced voting methods, production notification profiles, and reference accessibility evidence remain incomplete.",),
owned_concepts=("poll", "poll option", "poll invitation", "poll response"),
non_owned_concepts=("scheduling request", "calendar event", "evaluation rubric"),
recovery_docs=("README.md",),
security_docs=("README.md",),
),
)
def get_manifest() -> ModuleManifest:
return manifest