File size: 1,426 Bytes
a12d188 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 | """Torch-free application services shared between the model-heavy dependency layer
and the lightweight routers. Importing this module never pulls torch or the biometric
models, so the notification/OOB router (and its API tests) stay torch-free.
The `notifications` singleton IS the instance the full app uses (ModelState.notify
references it), so production and tests share one service + state. Persistence and
audit are optional hooks the full app wires at startup; they no-op when unwired
(e.g. in the light CI test app), matching the demo store's behavior."""
from __future__ import annotations
from typing import Callable, Optional
from amanpay.notifications.service import NotificationService
# Single shared notification/OOB service (referenced by ModelState.notify).
notifications = NotificationService()
_persist: Optional[Callable[[], None]] = None
_audit: Optional[Callable[[str, str, dict], None]] = None
def wire(persist: Optional[Callable[[], None]] = None,
audit: Optional[Callable[[str, str, dict], None]] = None) -> None:
"""Wire durable-persistence and audit callbacks from the full app (torch side)."""
global _persist, _audit
_persist = persist
_audit = audit
def persist_prefs() -> None:
if _persist is not None:
_persist()
def audit_event(user_id: str, action: str, detail: dict) -> None:
if _audit is not None:
_audit(user_id, action, detail)
|