63 lines
2.2 KiB
Python
63 lines
2.2 KiB
Python
from __future__ import annotations
|
|
|
|
from collections.abc import Mapping
|
|
|
|
from govoplan_core.core.modules import ModuleContext
|
|
from govoplan_core.core.notifications import NotificationDispatchProvider, NotificationDispatchRequest
|
|
from govoplan_notifications.backend.service import deliver_notification, deliver_pending, enqueue_dispatch_request, notification_response
|
|
|
|
|
|
class SqlNotificationDispatchProvider(NotificationDispatchProvider):
|
|
def __init__(self, context: ModuleContext) -> None:
|
|
self._settings = context.settings
|
|
self._registry = context.registry
|
|
|
|
def tenant_id_for_notification(
|
|
self,
|
|
session: object,
|
|
*,
|
|
notification_id: str,
|
|
) -> str | None:
|
|
from govoplan_notifications.backend.db.models import NotificationMessage
|
|
|
|
notification = session.get(NotificationMessage, notification_id) # type: ignore[attr-defined]
|
|
return notification.tenant_id if notification is not None else None
|
|
|
|
def enqueue_notification(
|
|
self,
|
|
session: object,
|
|
request: NotificationDispatchRequest,
|
|
*,
|
|
enqueue_delivery: bool = True,
|
|
) -> Mapping[str, object]:
|
|
notification = enqueue_dispatch_request(session, request, enqueue_delivery=enqueue_delivery) # type: ignore[arg-type]
|
|
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,
|
|
registry=self._registry,
|
|
) # type: ignore[arg-type]
|
|
return notification_response(notification)
|
|
|
|
def deliver_pending(
|
|
self,
|
|
session: object,
|
|
*,
|
|
tenant_id: str | None = None,
|
|
limit: int = 50,
|
|
) -> Mapping[str, object]:
|
|
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:
|
|
return SqlNotificationDispatchProvider(context)
|