Coverage for src/lexigram/notification/di/provider.py: 70%
115 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-26 02:32 +0800
1"""DI provider for lexigram-notification."""
3from __future__ import annotations
5import asyncio
6from typing import TYPE_CHECKING, Any
8from lexigram.contracts.core import HealthCheckResult, HealthStatus, ProviderPriority
9from lexigram.contracts.notification.protocols import (
10 PushChannelProtocol,
11 SMSChannelProtocol,
12)
13from lexigram.di.provider import Provider
14from lexigram.logging import get_logger
15from lexigram.notification.config import (
16 APNsDriverConfig,
17 FCMDriverConfig,
18 NamedPushConfig,
19 NamedSMSConfig,
20 NotificationConfig,
21 TwilioDriverConfig,
22 WebPushDriverConfig,
23)
25if TYPE_CHECKING:
26 from lexigram.contracts.core.di import (
27 ContainerRegistrarProtocol,
28 ContainerResolverProtocol,
29 )
31logger = get_logger(__name__)
33try:
34 from lexigram.notification.backends.sms.twilio import TwilioSMS
35except ImportError:
36 TwilioSMS = None # type: ignore[assignment,misc]
38try:
39 from lexigram.notification.backends.push.fcm import FCMPush
40except ImportError:
41 FCMPush = None # type: ignore[assignment,misc]
43try:
44 from lexigram.notification.backends.push.apns import APNsPush
45except ImportError:
46 APNsPush = None # type: ignore[assignment,misc]
48try:
49 from lexigram.notification.backends.push.web_push import WebPushChannel
50except ImportError:
51 WebPushChannel = None # type: ignore[assignment,misc]
54class NotificationProvider(Provider):
55 """Register SMS and push notification services into the DI container.
57 Reads :class:`~lexigram.notification.config.NotificationConfig`, creates
58 the appropriate backends, and registers them as ``SMSChannelProtocol`` and
59 ``PushChannelProtocol``.
61 Supports multi-backend (``NotificationConfig.sms_backends`` and
62 ``NotificationConfig.push_backends``) mode. Each entry is registered under
63 its name via ``container.singleton(name=entry.name)``. The primary backend
64 (``primary=True`` or the first entry) also receives the unnamed bindings
65 for backward compatibility.
66 """
68 name = "notification"
69 priority = ProviderPriority.INFRASTRUCTURE
70 config_key: str | None = "notification"
71 config_model: type | None = NotificationConfig
73 def __init__(self, config: NotificationConfig | None = None) -> None:
74 super().__init__()
75 self._requested_config = config
76 self._config = config or NotificationConfig()
77 self._sms_services: list[tuple[str, Any]] = []
78 self._push_services: list[tuple[str, Any]] = []
80 @classmethod
81 def from_config(
82 cls, config: NotificationConfig, **context: Any
83 ) -> NotificationProvider:
84 """Factory method for DI container setup."""
85 return cls(config)
87 def _create_sms(self, entry: NamedSMSConfig) -> Any:
88 """Instantiate the correct SMS implementation for a config."""
89 if entry.driver == "twilio":
90 if TwilioSMS is None:
91 raise ImportError("TwilioSMS unavailable")
92 cfg = entry.twilio or TwilioDriverConfig()
93 token = (
94 cfg.auth_token.get_secret_value()
95 if hasattr(cfg.auth_token, "get_secret_value")
96 and cfg.auth_token is not None
97 else (cfg.auth_token or "")
98 )
99 return TwilioSMS(
100 account_sid=cfg.account_sid or "",
101 auth_token=token or "",
102 from_number=cfg.from_number,
103 timeout=cfg.timeout,
104 )
105 raise ValueError(f"Unsupported SMS driver: {entry.driver!r}")
107 def _create_push(self, entry: NamedPushConfig) -> Any:
108 """Instantiate the correct push implementation for a config."""
109 if entry.driver == "fcm":
110 if FCMPush is None:
111 raise ImportError("FCMPush unavailable")
112 fcm_cfg = entry.fcm or FCMDriverConfig()
113 server_key = (
114 fcm_cfg.server_key.get_secret_value()
115 if hasattr(fcm_cfg.server_key, "get_secret_value")
116 and fcm_cfg.server_key is not None
117 else (fcm_cfg.server_key or "")
118 )
119 return FCMPush(server_key=server_key or "", timeout=fcm_cfg.timeout)
120 if entry.driver == "apns":
121 if APNsPush is None:
122 raise ImportError(
123 "APNsPush unavailable — install lexigram-notification[apns]"
124 )
125 apns_cfg = entry.apns or APNsDriverConfig()
126 return APNsPush(
127 team_id=apns_cfg.team_id or "",
128 key_id=apns_cfg.key_id or "",
129 apns_auth_key=(
130 apns_cfg.apns_auth_key.get_secret_value()
131 if hasattr(apns_cfg.apns_auth_key, "get_secret_value")
132 and apns_cfg.apns_auth_key is not None
133 else str(apns_cfg.apns_auth_key or "")
134 ),
135 bundle_id=apns_cfg.bundle_id or "",
136 sandbox=apns_cfg.sandbox,
137 timeout=apns_cfg.timeout,
138 )
139 if entry.driver == "web_push":
140 if WebPushChannel is None:
141 raise ImportError(
142 "WebPushChannel unavailable — install lexigram-notification[web-push]"
143 )
144 wp_cfg = entry.web_push or WebPushDriverConfig()
145 return WebPushChannel(
146 vapid_private_key=getattr(
147 wp_cfg.vapid_private_key,
148 "get_secret_value",
149 lambda: wp_cfg.vapid_private_key, # type: ignore[arg-type]
150 )(),
151 vapid_public_key=wp_cfg.vapid_public_key or "",
152 vapid_claims_subject=wp_cfg.vapid_claims_subject or "",
153 http_timeout=wp_cfg.timeout,
154 )
155 raise ValueError(f"Unsupported push driver: {entry.driver!r}")
157 async def register(self, container: ContainerRegistrarProtocol) -> None:
158 """Bind all SMS and push backends into the container."""
159 self._config = self._requested_config or (
160 self.config
161 if isinstance(getattr(self, "config", None), NotificationConfig)
162 else self._config
163 )
164 container.singleton(NotificationConfig, self._config)
166 for entry in self._config.sms_backends:
167 backend = self._create_sms(entry)
168 self._sms_services.append((entry.name, backend))
169 container.singleton(
170 SMSChannelProtocol,
171 factory=lambda *_, b=backend: b,
172 name=entry.name,
173 )
174 is_primary = entry.primary or (
175 not any(e.primary for e in self._config.sms_backends)
176 and self._config.sms_backends[0] is entry
177 )
178 if is_primary:
179 container.singleton(SMSChannelProtocol, factory=lambda *_, b=backend: b)
181 for entry in self._config.push_backends:
182 backend = self._create_push(entry)
183 self._push_services.append((entry.name, backend))
184 container.singleton(
185 PushChannelProtocol,
186 factory=lambda *_, b=backend: b,
187 name=entry.name,
188 )
189 is_primary = entry.primary or (
190 not any(e.primary for e in self._config.push_backends)
191 and self._config.push_backends[0] is entry
192 )
193 if is_primary:
194 container.singleton(
195 PushChannelProtocol, factory=lambda *_, b=backend: b
196 )
198 logger.info(
199 "notification_registered",
200 sms=[n for n, _ in self._sms_services],
201 push=[n for n, _ in self._push_services],
202 )
204 async def boot(self, container: ContainerResolverProtocol) -> None:
205 """Health-check all backends; log warnings for degraded ones."""
206 all_services = self._sms_services + self._push_services
207 if not all_services:
208 return
209 results = await asyncio.gather(
210 *[svc.health_check() for _, svc in all_services],
211 return_exceptions=True,
212 )
213 for (name, _), result in zip(all_services, results, strict=True):
214 if isinstance(result, Exception):
215 logger.warning(
216 "notification_boot_unhealthy", backend=name, error=str(result)
217 )
219 async def shutdown(self) -> None:
220 """Shutdown in reverse registration order."""
221 for name, _ in reversed(self._push_services + self._sms_services):
222 logger.info("notification_shutdown", backend=name)
223 self._sms_services.clear()
224 self._push_services.clear()
226 async def health_check(self, timeout: float = 5.0) -> HealthCheckResult:
227 """Aggregate health across all registered notification backends."""
228 all_services = self._sms_services + self._push_services
229 if not all_services:
230 return HealthCheckResult(
231 component="notification",
232 status=HealthStatus.HEALTHY,
233 details={"backends": []},
234 )
236 results = await asyncio.gather(
237 *[svc.health_check() for _, svc in all_services],
238 return_exceptions=True,
239 )
240 worst = HealthStatus.HEALTHY
241 details: dict[str, Any] = {}
242 for (name, _), result in zip(all_services, results, strict=True):
243 if isinstance(result, Exception):
244 worst = HealthStatus.UNHEALTHY
245 details[name] = {"status": "error", "error": str(result)}
246 elif isinstance(result, HealthCheckResult):
247 details[name] = {"status": result.status.value}
248 if result.status == HealthStatus.UNHEALTHY:
249 worst = HealthStatus.UNHEALTHY
250 elif (
251 result.status == HealthStatus.DEGRADED
252 and worst == HealthStatus.HEALTHY
253 ):
254 worst = HealthStatus.DEGRADED
256 return HealthCheckResult(
257 component="notification", status=worst, details=details
258 )
261__all__ = ["NotificationProvider"]