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

feat: route email notifications through mail capability

Browse Source
This commit is contained in:
zemion
2026-07-30 17:42:07 +02:00
parent e32ba3663b
commit 8153a1de45
9 changed files with 311 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/govoplan_notifications/backend/capabilities.py
+14 -2
View File
@@ -10,6 +10,7 @@ from govoplan_notifications.backend.service import deliver_notification, deliver
class SqlNotificationDispatchProvider(NotificationDispatchProvider):
def __init__(self, context: ModuleContext) -> None:
self._settings = context.settings
self._registry = context.registry
def enqueue_notification(
self,
@@ -22,7 +23,12 @@ class SqlNotificationDispatchProvider(NotificationDispatchProvider):
return notification_response(notification)
def deliver_notification(self, session: object, *, notification_id: str) -> Mapping[str, object]:
notification = deliver_notification(session, notification_id=notification_id, settings=self._settings) # type: ignore[arg-type]
notification = deliver_notification(
session,
notification_id=notification_id,
settings=self._settings,
registry=self._registry,
) # type: ignore[arg-type]
return notification_response(notification)
def deliver_pending(
@@ -32,7 +38,13 @@ class SqlNotificationDispatchProvider(NotificationDispatchProvider):
tenant_id: str | None = None,
limit: int = 50,
) -> Mapping[str, object]:
return deliver_pending(session, tenant_id=tenant_id, limit=limit, settings=self._settings) # type: ignore[arg-type]
return deliver_pending(
session,
tenant_id=tenant_id,
limit=limit,
settings=self._settings,
registry=self._registry,
) # type: ignore[arg-type]
def dispatch_capability(context: ModuleContext) -> SqlNotificationDispatchProvider:
src/govoplan_notifications/backend/router.py
+19 -2
View File
@@ -6,7 +6,9 @@ from fastapi import APIRouter, Depends, HTTPException, Query, status
from sqlalchemy.orm import Session
from govoplan_core.auth import ApiPrincipal, get_api_principal, has_scope
from govoplan_core.core.runtime import get_registry
from govoplan_core.db.session import get_session
from govoplan_core.settings import settings as core_settings
from govoplan_notifications.backend.manifest import ADMIN_SCOPE, DISPATCH_SCOPE, READ_SCOPE, WRITE_SCOPE
from govoplan_notifications.backend.schemas import (
NotificationCreateRequest,
@@ -169,7 +171,13 @@ def api_deliver_pending(
_require_scope(principal, DISPATCH_SCOPE)
if payload.tenant_id is not None:
_require_scope(principal, ADMIN_SCOPE)
result = deliver_pending(session, tenant_id=payload.tenant_id or principal.tenant_id, limit=payload.limit)
result = deliver_pending(
session,
tenant_id=payload.tenant_id or principal.tenant_id,
limit=payload.limit,
settings=core_settings,
registry=get_registry(),
)
session.commit()
return NotificationDeliveryResultResponse.model_validate(result)
@@ -227,17 +235,26 @@ def api_deliver_notification(
_require_scope(principal, DISPATCH_SCOPE)
try:
get_notification(session, tenant_id=principal.tenant_id, notification_id=notification_id)
notification = deliver_notification(session, notification_id=notification_id)
notification = deliver_notification(
session,
notification_id=notification_id,
settings=core_settings,
registry=get_registry(),
)
except NotificationError as exc:
raise _notification_http_error(exc) from exc
session.commit()
sent = 1 if notification.status == "sent" else 0
failed = 1 if notification.status == "failed" else 0
skipped = 1 if notification.status == "skipped" else 0
accepted = 1 if notification.status == "accepted" else 0
paused = 1 if notification.status == "paused" else 0
return NotificationDeliveryResultResponse(
notification=_response(notification),
processed=1,
sent=sent,
accepted=accepted,
paused=paused,
failed=failed,
skipped=skipped,
errors=[notification.last_error] if notification.last_error else [],
src/govoplan_notifications/backend/schemas.py
+13 -1
View File
@@ -6,7 +6,17 @@ from typing import Any, Literal
from pydantic import BaseModel, ConfigDict, Field, field_validator
NotificationStatus = Literal["pending", "queued", "sending", "sent", "failed", "skipped", "cancelled"]
NotificationStatus = Literal[
"pending",
"queued",
"sending",
"accepted",
"paused",
"sent",
"failed",
"skipped",
"cancelled",
]
def normalize_notification_action_url(value: str | None) -> str | None:
@@ -154,6 +164,8 @@ class NotificationDeliveryResultResponse(BaseModel):
notification: NotificationResponse | None = None
processed: int = 0
sent: int = 0
accepted: int = 0
paused: int = 0
failed: int = 0
skipped: int = 0
errors: list[str] = Field(default_factory=list)
src/govoplan_notifications/backend/service.py
+109 -9
View File
@@ -9,7 +9,12 @@ from typing import Any
from sqlalchemy import and_, case, event, func, or_
from sqlalchemy.orm import Session
from govoplan_core.core.mail import (
NotificationMailDeliveryRequest,
notification_mail_delivery_provider,
)
from govoplan_core.core.notifications import NotificationDispatchRequest
from govoplan_core.core.runtime import get_registry
from govoplan_core.db.base import utcnow
from govoplan_notifications.backend.db.models import NotificationDeliveryAttempt, NotificationMessage, NotificationPreference
from govoplan_notifications.backend.schemas import (
@@ -25,6 +30,7 @@ class NotificationError(ValueError):
PENDING_STATUSES = {"pending", "queued", "failed"}
SUMMARY_PENDING_STATUSES = PENDING_STATUSES | {"accepted", "paused", "sending"}
_AFTER_COMMIT_DELIVERY_IDS = "govoplan_notifications_after_commit_delivery_ids"
@@ -232,7 +238,7 @@ def notification_summary(
and_(
active,
NotificationMessage.status.in_(
sorted(PENDING_STATUSES)
sorted(SUMMARY_PENDING_STATUSES)
),
),
1,
@@ -329,6 +335,7 @@ def deliver_notification(
*,
notification_id: str,
settings: object | None = None,
registry: object | None = None,
) -> NotificationMessage:
# An explicit same-transaction delivery supersedes the deferred Celery
# handoff registered when the row was created.
@@ -336,7 +343,7 @@ def deliver_notification(
notification = session.get(NotificationMessage, notification_id)
if notification is None or notification.deleted_at is not None:
raise NotificationError("Notification not found")
if notification.status in {"sent", "skipped", "cancelled"}:
if notification.status in {"accepted", "sent", "skipped", "cancelled"}:
return notification
if notification.not_before_at is not None and response_datetime(notification.not_before_at) > _now():
notification.status = "queued"
@@ -345,7 +352,12 @@ def deliver_notification(
attempt = _start_attempt(notification)
try:
result = _deliver_by_channel(notification, settings=settings)
result = _deliver_by_channel(
session,
notification,
settings=settings,
registry=registry if registry is not None else get_registry(),
)
except Exception as exc: # noqa: BLE001 - persisted as delivery failure.
_finish_attempt(attempt, status="failed", error=str(exc))
notification.status = "failed"
@@ -353,7 +365,13 @@ def deliver_notification(
notification.last_error = str(exc)
session.flush()
return notification
_finish_attempt(attempt, status=result["status"], provider=result.get("provider"), details=result)
_finish_attempt(
attempt,
status=result["status"],
provider=result.get("provider"),
error=result.get("error"),
details=result,
)
notification.status = result["status"]
notification.sent_at = _now() if result["status"] == "sent" else notification.sent_at
notification.failed_at = _now() if result["status"] == "failed" else notification.failed_at
@@ -369,6 +387,7 @@ def deliver_pending(
tenant_id: str | None = None,
limit: int = 50,
settings: object | None = None,
registry: object | None = None,
) -> dict[str, Any]:
now = _now()
query = session.query(NotificationMessage).filter(
@@ -379,12 +398,31 @@ def deliver_pending(
if tenant_id:
query = query.filter(NotificationMessage.tenant_id == tenant_id)
notifications = query.order_by(NotificationMessage.priority.desc(), NotificationMessage.created_at.asc()).limit(limit).all()
result = {"processed": 0, "sent": 0, "failed": 0, "skipped": 0, "errors": []}
result = {
"processed": 0,
"sent": 0,
"accepted": 0,
"paused": 0,
"failed": 0,
"skipped": 0,
"errors": [],
}
for notification in notifications:
result["processed"] += 1
delivered = deliver_notification(session, notification_id=notification.id, settings=settings)
delivered = deliver_notification(
session,
notification_id=notification.id,
settings=settings,
registry=registry,
)
if delivered.status == "sent":
result["sent"] += 1
elif delivered.status == "accepted":
result["accepted"] += 1
elif delivered.status == "paused":
result["paused"] += 1
if delivered.last_error:
result["errors"].append(delivered.last_error)
elif delivered.status == "skipped":
result["skipped"] += 1
elif delivered.status == "failed":
@@ -496,17 +534,79 @@ def _finish_attempt(
attempt.external_message_id = str(details["external_message_id"])
def _deliver_by_channel(notification: NotificationMessage, *, settings: object | None) -> dict[str, Any]:
def _deliver_by_channel(
session: Session,
notification: NotificationMessage,
*,
settings: object | None,
registry: object | None,
) -> dict[str, Any]:
if notification.channel == "inbox":
return {"status": "sent", "provider": "inbox", "external_message_id": notification.id}
if notification.channel == "mail":
if not notification.recipient:
return {"status": "skipped", "provider": "mail", "error": "Mail notification has no recipient address"}
path = _write_local_mail(notification, settings=settings)
return {"status": "sent", "provider": "local_file_mail", "external_message_id": str(path), "path": str(path)}
provider = notification_mail_delivery_provider(registry)
if provider is not None:
mail_settings = _notification_mail_settings(notification)
return dict(
provider.submit_notification_mail(
session,
NotificationMailDeliveryRequest(
tenant_id=notification.tenant_id,
notification_id=notification.id,
recipient=notification.recipient,
subject=notification.subject or f"Notification: {notification.event_kind}",
body_text=notification.body_text or notification.subject or notification.event_kind,
body_html=notification.body_html,
action_url=notification.action_url,
mail_profile_id=_optional_text(mail_settings.get("mail_profile_id")),
from_address=_optional_text(mail_settings.get("from_address")),
smtp_server_id=_optional_text(mail_settings.get("smtp_server_id")),
smtp_credential_id=_optional_text(mail_settings.get("smtp_credential_id")),
metadata={
"source_module": notification.source_module,
"event_kind": notification.event_kind,
},
),
)
)
if _development_file_mail_enabled(settings):
path = _write_local_mail(notification, settings=settings)
return {
"status": "sent",
"provider": "local_file_mail",
"external_message_id": str(path),
"path": str(path),
}
return {
"status": "paused",
"provider": "mail",
"error": "Mail-backed notification delivery is unavailable.",
}
return {"status": "skipped", "provider": notification.channel, "error": f"No delivery adapter configured for channel {notification.channel!r}"}
def _notification_mail_settings(
notification: NotificationMessage,
) -> dict[str, object]:
metadata = notification.metadata_ if isinstance(notification.metadata_, dict) else {}
value = metadata.get("mail_delivery")
return dict(value) if isinstance(value, dict) else {}
def _optional_text(value: object) -> str | None:
text = str(value or "").strip()
return text or None
def _development_file_mail_enabled(settings: object | None) -> bool:
if settings is None:
return False
app_env = str(getattr(settings, "app_env", "")).strip().lower()
return app_env in {"dev", "development", "test"}
def _write_local_mail(notification: NotificationMessage, *, settings: object | None) -> Path:
root = Path(str(getattr(settings, "mock_mailbox_dir", "runtime/mock-mailbox"))) / "notifications"
root.mkdir(parents=True, exist_ok=True)