add-ideas/govoplan-scheduling
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

intermittent commit

Browse Source
This commit is contained in:
Albrecht Degering
2026-07-14 13:22:12 +02:00
parent c1afce7bdb
commit 2f1b7fb6b8
9 changed files with 364 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
src/govoplan_scheduling/backend/manifest.py
View File

@@ -114,9 +114,11 @@ def _tenant_summary(session, tenant_id: str) -> dict[str, int]:
}
def _scheduling_router(_context: ModuleContext):
def _scheduling_router(context: ModuleContext):
from govoplan_scheduling.backend.runtime import configure_runtime
from govoplan_scheduling.backend.router import router
configure_runtime(registry=context.registry, settings=context.settings)
return router
@@ -125,7 +127,7 @@ manifest = ModuleManifest(
name=MODULE_NAME,
version=MODULE_VERSION,
dependencies=("poll",),
optional_dependencies=("access", "calendar", "appointments", "evaluation", "mail", "notifications", "portal", "workflow", "tasks", "idm", "organizations"),
optional_dependencies=("access", "calendar", "appointments", "evaluation", "mail", "notifications", "portal", "workflow", "tasks", "idm", "organizations", "addresses"),
optional_capabilities=(CAPABILITY_AUTH_PRINCIPAL_RESOLVER, CAPABILITY_AUTH_PERMISSION_EVALUATOR),
provides_interfaces=(
ModuleInterfaceProvider(name="scheduling.candidate_slots", version="0.1.8"),
@@ -136,14 +138,16 @@ manifest = ModuleManifest(
ModuleInterfaceRequirement(name="poll.workflow_context", version_min="0.1.8", version_max_exclusive="0.2.0"),
ModuleInterfaceRequirement(name="poll.signed_participation", version_min="0.1.8", version_max_exclusive="0.2.0", optional=True),
ModuleInterfaceRequirement(name="evaluation.feedback", version_min="0.1.8", version_max_exclusive="0.2.0", optional=True),
ModuleInterfaceRequirement(name="notifications.dispatch", version_min="0.1.8", version_max_exclusive="0.2.0", optional=True),
ModuleInterfaceRequirement(name="addresses.lookup", version_min="0.1.0", version_max_exclusive="0.2.0", optional=True),
),
permissions=PERMISSIONS,
role_templates=ROLE_TEMPLATES,
nav_items=(NavItem(path="/scheduling", label="Scheduling", icon="calendar", required_any=(READ_SCOPE,), order=56),),
nav_items=(NavItem(path="/scheduling", label="Scheduling", icon="calendar-clock", required_any=(READ_SCOPE,), order=56),),
frontend=FrontendModule(
module_id=MODULE_ID,
package_name="@govoplan/scheduling-webui",
nav_items=(NavItem(path="/scheduling", label="Scheduling", icon="calendar", required_any=(READ_SCOPE,), order=56),),
nav_items=(NavItem(path="/scheduling", label="Scheduling", icon="calendar-clock", required_any=(READ_SCOPE,), order=56),),
),
route_factory=_scheduling_router,
tenant_summary_providers=(_tenant_summary,),

57
src/govoplan_scheduling/backend/router.py
View File

@@ -1,5 +1,8 @@
from __future__ import annotations
import dataclasses
from typing import Any
from fastapi import APIRouter, Depends, HTTPException, Query, status
from sqlalchemy.orm import Session
@@ -8,6 +11,8 @@ from govoplan_core.db.session import get_session
from govoplan_poll.backend.schemas import PollResultSummaryResponse
from govoplan_scheduling.backend.manifest import READ_SCOPE, WRITE_SCOPE
from govoplan_scheduling.backend.schemas import (
SchedulingAddressLookupCandidate,
SchedulingAddressLookupResponse,
SchedulingCalendarActionResponse,
SchedulingDecisionRequest,
SchedulingNotificationCreateRequest,
@@ -20,6 +25,7 @@ from govoplan_scheduling.backend.schemas import (
SchedulingStatusResponse,
SchedulingSummaryResponse,
)
from govoplan_scheduling.backend.runtime import get_registry
from govoplan_scheduling.backend.service import (
SchedulingError,
cancel_scheduling_request,
@@ -42,6 +48,39 @@ from govoplan_scheduling.backend.service import (
router = APIRouter(prefix="/scheduling", tags=["scheduling"])
CAPABILITY_ADDRESSES_LOOKUP = "addresses.lookup"
def _capability_payload(value: object) -> dict[str, Any]:
if dataclasses.is_dataclass(value):
return dataclasses.asdict(value)
if isinstance(value, dict):
return dict(value)
payload: dict[str, Any] = {}
for key in (
"contact_id",
"address_book_id",
"display_name",
"email",
"email_label",
"organization",
"role_title",
"tags",
"source_kind",
"source_ref",
"source_revision",
"provenance",
):
if hasattr(value, key):
payload[key] = getattr(value, key)
return payload
def _registry_capability(name: str) -> object | None:
registry = get_registry()
if registry is None or not hasattr(registry, "has_capability") or not registry.has_capability(name):
return None
return registry.capability(name)
def _require_scope(principal: ApiPrincipal, scope: str) -> None:
@@ -61,6 +100,24 @@ def _request_response(request, *, invitation_tokens: dict[str, str] | None = Non
)
@router.get("/address-lookup", response_model=SchedulingAddressLookupResponse)
def api_lookup_scheduling_addresses(
query: str = Query(min_length=1),
limit: int = Query(default=25, ge=1, le=100),
session: Session = Depends(get_session),
principal: ApiPrincipal = Depends(get_api_principal),
) -> SchedulingAddressLookupResponse:
_require_scope(principal, WRITE_SCOPE)
capability = _registry_capability(CAPABILITY_ADDRESSES_LOOKUP)
if capability is None or not hasattr(capability, "lookup"):
return SchedulingAddressLookupResponse(available=False, candidates=[])
candidates = getattr(capability, "lookup")(session, principal, query=query, limit=limit)
return SchedulingAddressLookupResponse(
available=True,
candidates=[SchedulingAddressLookupCandidate.model_validate(_capability_payload(candidate)) for candidate in candidates],
)
@router.get("/requests", response_model=SchedulingRequestListResponse)
def api_list_scheduling_requests(
status_filter: str | None = Query(default=None, alias="status"),

20
src/govoplan_scheduling/backend/runtime.py Normal file
View File

@@ -0,0 +1,20 @@
from __future__ import annotations
_runtime_registry: object | None = None
_runtime_settings: object | None = None
def configure_runtime(*, registry: object | None = None, settings: object | None = None) -> None:
global _runtime_registry, _runtime_settings
if registry is not None:
_runtime_registry = registry
if settings is not None:
_runtime_settings = settings
def get_registry() -> object | None:
return _runtime_registry
def get_settings() -> object | None:
return _runtime_settings

20
src/govoplan_scheduling/backend/schemas.py
View File

@@ -202,3 +202,23 @@ class SchedulingNotificationCreateRequest(BaseModel):
event_kind: Literal["invitation", "reminder", "decision", "cancellation"]
channel: str = Field(default="mail", max_length=40)
metadata: dict[str, Any] = Field(default_factory=dict)
class SchedulingAddressLookupCandidate(BaseModel):
contact_id: str
address_book_id: str
display_name: str
email: str | None = None
email_label: str | None = None
organization: str | None = None
role_title: str | None = None
tags: list[str] = Field(default_factory=list)
source_kind: str = "local"
source_ref: str | None = None
source_revision: str | None = None
provenance: dict[str, Any] = Field(default_factory=dict)
class SchedulingAddressLookupResponse(BaseModel):
available: bool = False
candidates: list[SchedulingAddressLookupCandidate] = Field(default_factory=list)

175
src/govoplan_scheduling/backend/service.py
View File

@@ -5,6 +5,7 @@ from typing import Any
from sqlalchemy.orm import Session
from govoplan_core.core.notifications import NotificationDispatchRequest, notification_dispatch_provider
from govoplan_core.db.base import utcnow
from govoplan_poll.backend.schemas import PollCreateRequest, PollDecisionRequest, PollInvitationCreateRequest, PollOptionInput
from govoplan_poll.backend.db.models import PollResponse
@@ -24,6 +25,7 @@ from govoplan_scheduling.backend.schemas import (
SchedulingRequestCreateRequest,
SchedulingRequestUpdateRequest,
)
from govoplan_scheduling.backend.runtime import get_registry
class SchedulingError(ValueError):
@@ -348,6 +350,7 @@ def create_scheduling_notification_jobs(
raise SchedulingError(f"Unsupported scheduling notification kind: {event_kind}")
request = get_scheduling_request(session, tenant_id=tenant_id, request_id=request_id)
notifications: list[SchedulingNotification] = []
base_metadata = metadata or {}
for participant in _active_participants(request):
recipient = participant.email or participant.respondent_id
status = "pending" if recipient else "skipped"
@@ -367,16 +370,178 @@ def create_scheduling_notification_jobs(
"selected_slot_id": request.selected_slot_id,
"calendar_event_id": request.calendar_event_id,
},
metadata_=metadata or {},
metadata_=dict(base_metadata),
)
if status == "skipped":
notification.error = "Participant has no deliverable recipient address or id"
session.add(notification)
notifications.append(notification)
session.flush()
_mirror_scheduling_notifications_to_center(session, request=request, notifications=notifications)
return notifications
def _mirror_scheduling_notifications_to_center(
session: Session,
*,
request: SchedulingRequest,
notifications: list[SchedulingNotification],
) -> None:
provider = notification_dispatch_provider(get_registry())
if provider is None:
return
participants_by_id = {participant.id: participant for participant in _active_participants(request)}
for notification in notifications:
if notification.status == "skipped":
continue
participant = participants_by_id.get(notification.participant_id or "")
try:
response = provider.enqueue_notification(
session,
_notification_dispatch_request(request=request, participant=participant, notification=notification),
enqueue_delivery=True,
)
except Exception as exc: # noqa: BLE001 - mirroring must not block scheduling workflow.
notification.status = "failed"
notification.error = f"Notification center enqueue failed: {exc}"
continue
notification.metadata_ = {
**(notification.metadata_ or {}),
"notification_id": response.get("id"),
"notification_status": response.get("status"),
}
if response.get("status") in {"queued", "sending", "sent"}:
notification.status = str(response["status"])
elif response.get("status") == "skipped":
notification.status = "skipped"
notification.error = str(response.get("last_error") or "Notification center skipped delivery")
session.flush()
def _notification_dispatch_request(
*,
request: SchedulingRequest,
participant: SchedulingParticipant | None,
notification: SchedulingNotification,
) -> NotificationDispatchRequest:
payload = {
**(notification.payload or {}),
"scheduling_notification_id": notification.id,
"participant": {
"id": participant.id,
"display_name": participant.display_name,
"email": participant.email,
"required": participant.required,
"participant_type": participant.participant_type,
}
if participant
else None,
}
return NotificationDispatchRequest(
tenant_id=request.tenant_id,
source_module="scheduling",
source_resource_type="request",
source_resource_id=request.id,
event_kind=notification.event_kind,
channel=notification.channel,
recipient=notification.recipient,
recipient_type="scheduling_participant",
recipient_id=participant.id if participant else notification.participant_id,
recipient_label=participant.display_name if participant else None,
subject=_notification_subject(request, notification.event_kind),
body_text=_notification_body_text(request, participant, notification.event_kind),
action_url=f"/scheduling?request_id={request.id}",
payload=_jsonable(payload),
metadata={
**(notification.metadata_ or {}),
"scheduling_notification_id": notification.id,
"scheduling_request_id": request.id,
},
)
def _notification_subject(request: SchedulingRequest, event_kind: str) -> str:
labels = {
"invitation": "Scheduling invitation",
"reminder": "Scheduling reminder",
"decision": "Scheduling decision",
"cancellation": "Scheduling cancelled",
}
return f"{labels.get(event_kind, 'Scheduling update')}: {request.title}"
def _notification_body_text(
request: SchedulingRequest,
participant: SchedulingParticipant | None,
event_kind: str,
) -> str:
greeting = f"Hello {participant.display_name}," if participant and participant.display_name else "Hello,"
lines = [greeting, "", _notification_body_intro(request, event_kind)]
selected_slot = next((slot for slot in _active_slots(request) if slot.id == request.selected_slot_id), None)
if selected_slot is not None:
lines.append(f"Selected slot: {_slot_label(selected_slot.start_at, selected_slot.end_at, timezone_name=selected_slot.timezone)}")
if request.location:
lines.append(f"Location: {request.location}")
lines.append(f"Scheduling request: {request.title}")
return "\n".join(lines)
def _notification_body_intro(request: SchedulingRequest, event_kind: str) -> str:
if event_kind == "invitation":
return f"You are invited to respond to the scheduling poll for {request.title}."
if event_kind == "reminder":
return f"This is a reminder to respond to the scheduling poll for {request.title}."
if event_kind == "decision":
return f"A final slot was selected for {request.title}."
if event_kind == "cancellation":
return f"The scheduling request {request.title} was cancelled."
return f"There is an update for {request.title}."
def _emit_scheduling_center_notification(
session: Session,
*,
request: SchedulingRequest,
event_kind: str,
subject: str,
body_text: str,
participant: SchedulingParticipant | None = None,
priority: int = 0,
) -> None:
provider = notification_dispatch_provider(get_registry())
if provider is None:
return
try:
provider.enqueue_notification(
session,
NotificationDispatchRequest(
tenant_id=request.tenant_id,
source_module="scheduling",
source_resource_type="request",
source_resource_id=request.id,
event_kind=event_kind,
channel="inbox",
recipient_type="user" if request.organizer_user_id else None,
recipient_id=request.organizer_user_id,
subject=subject,
body_text=body_text,
action_url=f"/scheduling?request_id={request.id}",
priority=priority,
payload=_jsonable(
{
"request_id": request.id,
"poll_id": request.poll_id,
"participant_id": participant.id if participant else None,
}
),
metadata={"scheduling_request_id": request.id},
),
enqueue_delivery=False,
)
except Exception:
return
def list_scheduling_notifications(
session: Session,
*,
@@ -414,6 +579,14 @@ def refresh_participant_response_state(session: Session, *, request: SchedulingR
if participant is not None and participant.status != "responded":
participant.status = "responded"
participant.responded_at = response_datetime(response.submitted_at)
_emit_scheduling_center_notification(
session,
request=request,
participant=participant,
event_kind="scheduling.participant_responded",
subject=f"Scheduling response: {request.title}",
body_text=f"{participant.display_name or participant.email or 'A participant'} responded to {request.title}.",
)
changed = True
if changed:
session.flush()