Coverage for src / lexigram / admin / services / notifications / service.py: 28%

1"""Admin notification service - main orchestration class.""" 

2 

3from __future__ import annotations 

4 

5from datetime import UTC, datetime 

6from typing import TYPE_CHECKING, Any 

7 

8from lexigram.admin.config import AdminNotificationConfig 

9from lexigram.admin.exceptions import NotificationError 

10from lexigram.admin.services.notifications.models import ( 

11 Notification, 

12 NotificationRecipient, 

13 NotificationResult, 

14 NotificationType, 

15) 

16from lexigram.admin.services.notifications.sender import EmailSender 

17from lexigram.admin.services.notifications.templates import TemplateRenderer 

18from lexigram.di.decorators import inject 

19from lexigram.result import Err, Ok, Result 

20 

21if TYPE_CHECKING: 

22 from lexigram.contracts.mailer.protocols import MailerProtocol 

23 

24 

25@inject 

26class AdminNotificationService: 

27 """Service for sending admin notifications. 

28 

29 Integrates with lexigram.messaging for email delivery 

30 and provides admin-specific templates and functionality. 

31 

32 Example: 

33 >>> service = AdminNotificationService(messaging, config) 

34 >>> 

35 >>> # Send user created notification 

36 >>> await service.notify_user_created( 

37 ... user=user, 

38 ... created_by=admin, 

39 ... recipients=admin_list, 

40 ... ) 

41 >>> 

42 >>> # Send bulk completion notification 

43 >>> await service.notify_bulk_completed( 

44 ... operation="delete", 

45 ... resource="users", 

46 ... total=100, 

47 ... successful=98, 

48 ... failed=2, 

49 ... recipients=[admin], 

50 ... ) 

51 """ 

52 

53 def __init__( 

54 self, 

55 mailer: MailerProtocol | None = None, 

56 config: AdminNotificationConfig | None = None, 

57 ): 

58 self.config = config or AdminNotificationConfig() 

59 

60 # Initialize components 

61 self.template_renderer = TemplateRenderer(self.config.app_name) # type: ignore[attr-defined] 

62 self.email_sender = EmailSender( 

63 mailer=mailer, 

64 from_email=self.config.from_email, # type: ignore[attr-defined] 

65 from_name=self.config.from_name, # type: ignore[attr-defined] 

66 ) 

67 

68 self._sent_count = 0 

69 

70 async def send( 

71 self, 

72 notification: Notification, 

73 ) -> Result[NotificationResult, NotificationError]: 

74 """Send a notification. 

75 

76 Args: 

77 notification: Notification to send 

78 

79 Returns: 

80 ``Ok(NotificationResult)`` on success or partial success. 

81 ``Err(NotificationError)`` when all recipients failed. 

82 """ 

83 if ( 

84 notification.type not in self.config.enabled_types # type: ignore[attr-defined] 

85 and self.config.enabled_types # type: ignore[attr-defined] 

86 ): 

87 result = NotificationResult( 

88 notification_id=notification.id, 

89 recipients_sent=0, 

90 errors=["Notification type not enabled"], 

91 ) 

92 return Ok(result) 

93 

94 sent = 0 

95 failed = 0 

96 errors: list[str] = [] 

97 

98 for recipient in notification.recipients: 

99 # Check recipient preferences 

100 if not recipient.can_receive(notification.type): 

101 continue 

102 

103 # Send via channels 

104 for channel in notification.channels: 

105 if channel.value == "email": # Use .value to compare with string 

106 try: 

107 await self.email_sender.send_email( 

108 recipient=recipient, 

109 subject=notification.subject, 

110 body=notification.body, 

111 html_body=notification.html_body, 

112 ) 

113 sent += 1 

114 except (RuntimeError, OSError, ConnectionError) as e: 

115 failed += 1 

116 errors.append(f"Email to {recipient.email}: {e}") 

117 

118 result = NotificationResult( 

119 notification_id=notification.id, 

120 recipients_sent=sent, 

121 recipients_failed=failed, 

122 errors=errors, 

123 ) 

124 

125 if sent == 0 and failed > 0: 

126 return Err( 

127 NotificationError( 

128 f"All {failed} recipient(s) failed: {'; '.join(errors)}" 

129 ) 

130 ) 

131 

132 return Ok(result) 

133 

134 # ======================================================================== 

135 # Convenience Methods 

136 # ======================================================================== 

137 

138 async def notify_user_created( 

139 self, 

140 user: Any, 

141 created_by: Any, 

142 recipients: list[NotificationRecipient], 

143 ) -> Result[NotificationResult, NotificationError]: 

144 """Send user created notification.""" 

145 data = { 

146 "user_name": getattr(user, "name", str(user)), 

147 "user_email": getattr(user, "email", ""), 

148 "user_role": getattr(user, "role", "User"), 

149 "created_by": getattr(created_by, "name", str(created_by)), 

150 "created_at": datetime.now(UTC).isoformat(), 

151 "user_url": f"{self.config.base_url}/users/{getattr(user, 'id', '')}", # type: ignore[attr-defined] 

152 } 

153 

154 subject, body, html_body = self.template_renderer.render_template( 

155 NotificationType.USER_CREATED, 

156 data, 

157 ) 

158 

159 notification = Notification( 

160 type=NotificationType.USER_CREATED, 

161 subject=subject, 

162 body=body, 

163 html_body=html_body, 

164 recipients=recipients, 

165 data=data, 

166 ) 

167 

168 return await self.send(notification) 

169 

170 async def notify_user_invited( 

171 self, 

172 user_email: str, 

173 user_name: str, 

174 invite_url: str, 

175 expires_in: str = "7 days", 

176 ) -> Result[NotificationResult, NotificationError]: 

177 """Send user invitation notification.""" 

178 data = { 

179 "user_name": user_name, 

180 "user_email": user_email, 

181 "invite_url": invite_url, 

182 "expires_in": expires_in, 

183 } 

184 

185 subject, body, html_body = self.template_renderer.render_template( 

186 NotificationType.USER_INVITED, 

187 data, 

188 ) 

189 

190 recipient = NotificationRecipient(email=user_email, name=user_name) 

191 

192 notification = Notification( 

193 type=NotificationType.USER_INVITED, 

194 subject=subject, 

195 body=body, 

196 html_body=html_body, 

197 recipients=[recipient], 

198 data=data, 

199 ) 

200 

201 return await self.send(notification) 

202 

203 async def notify_password_reset( 

204 self, 

205 user_email: str, 

206 user_name: str, 

207 reset_url: str, 

208 expires_in: str = "1 hour", 

209 ) -> Result[NotificationResult, NotificationError]: 

210 """Send password reset notification.""" 

211 data = { 

212 "user_name": user_name, 

213 "reset_url": reset_url, 

214 "expires_in": expires_in, 

215 } 

216 

217 subject, body, html_body = self.template_renderer.render_template( 

218 NotificationType.PASSWORD_RESET, 

219 data, 

220 ) 

221 

222 recipient = NotificationRecipient(email=user_email, name=user_name) 

223 

224 notification = Notification( 

225 type=NotificationType.PASSWORD_RESET, 

226 subject=subject, 

227 body=body, 

228 html_body=html_body, 

229 recipients=[recipient], 

230 data=data, 

231 ) 

232 

233 return await self.send(notification) 

234 

235 async def notify_bulk_started( 

236 self, 

237 operation_name: str, 

238 resource: str, 

239 total_items: int, 

240 started_by: Any, 

241 recipients: list[NotificationRecipient], 

242 ) -> Result[NotificationResult, NotificationError]: 

243 """Send bulk operation started notification.""" 

244 data = { 

245 "operation_name": operation_name, 

246 "resource": resource, 

247 "total_items": total_items, 

248 "started_by": getattr(started_by, "name", str(started_by)), 

249 "started_at": datetime.now(UTC).isoformat(), 

250 } 

251 

252 subject, body, html_body = self.template_renderer.render_template( 

253 NotificationType.BULK_STARTED, 

254 data, 

255 ) 

256 

257 notification = Notification( 

258 type=NotificationType.BULK_STARTED, 

259 subject=subject, 

260 body=body, 

261 html_body=html_body, 

262 recipients=recipients, 

263 data=data, 

264 ) 

265 

266 return await self.send(notification) 

267 

268 async def notify_bulk_completed( 

269 self, 

270 operation_name: str, 

271 resource: str, 

272 total_items: int, 

273 successful: int, 

274 failed: int, 

275 duration: str, 

276 recipients: list[NotificationRecipient], 

277 results_url: str | None = None, 

278 ) -> Result[NotificationResult, NotificationError]: 

279 """Send bulk operation completed notification.""" 

280 data = { 

281 "operation_name": operation_name, 

282 "resource": resource, 

283 "total_items": total_items, 

284 "successful": successful, 

285 "failed": failed, 

286 "duration": duration, 

287 "results_url": results_url or f"{self.config.base_url}/{resource}", # type: ignore[attr-defined] 

288 } 

289 

290 subject, body, html_body = self.template_renderer.render_template( 

291 NotificationType.BULK_COMPLETED, 

292 data, 

293 ) 

294 

295 notification = Notification( 

296 type=NotificationType.BULK_COMPLETED, 

297 subject=subject, 

298 body=body, 

299 html_body=html_body, 

300 recipients=recipients, 

301 data=data, 

302 ) 

303 

304 return await self.send(notification) 

305 

306 async def notify_bulk_failed( 

307 self, 

308 operation_name: str, 

309 resource: str, 

310 error_message: str, 

311 processed: int, 

312 recipients: list[NotificationRecipient], 

313 ) -> Result[NotificationResult, NotificationError]: 

314 """Send bulk operation failed notification.""" 

315 data = { 

316 "operation_name": operation_name, 

317 "resource": resource, 

318 "error_message": error_message, 

319 "processed": processed, 

320 } 

321 

322 subject, body, html_body = self.template_renderer.render_template( 

323 NotificationType.BULK_FAILED, 

324 data, 

325 ) 

326 

327 notification = Notification( 

328 type=NotificationType.BULK_FAILED, 

329 subject=subject, 

330 body=body, 

331 html_body=html_body, 

332 recipients=recipients, 

333 data=data, 

334 ) 

335 

336 return await self.send(notification) 

337 

338 async def notify_export_ready( 

339 self, 

340 export_name: str, 

341 file_format: str, 

342 record_count: int, 

343 file_size: str, 

344 download_url: str, 

345 recipient: NotificationRecipient, 

346 expires_in: str = "24 hours", 

347 ) -> Result[NotificationResult, NotificationError]: 

348 """Send export ready notification.""" 

349 data = { 

350 "export_name": export_name, 

351 "format": file_format, 

352 "record_count": record_count, 

353 "file_size": file_size, 

354 "download_url": download_url, 

355 "expires_in": expires_in, 

356 } 

357 

358 subject, body, html_body = self.template_renderer.render_template( 

359 NotificationType.EXPORT_READY, 

360 data, 

361 ) 

362 

363 notification = Notification( 

364 type=NotificationType.EXPORT_READY, 

365 subject=subject, 

366 body=body, 

367 html_body=html_body, 

368 recipients=[recipient], 

369 data=data, 

370 ) 

371 

372 return await self.send(notification) 

373 

374 async def notify_system_alert( 

375 self, 

376 alert_title: str, 

377 alert_message: str, 

378 severity: str, 

379 component: str, 

380 recipients: list[NotificationRecipient], 

381 ) -> Result[NotificationResult, NotificationError]: 

382 """Send system alert notification.""" 

383 severity_map = { 

384 "info": "info", 

385 "warning": "warning", 

386 "error": "error", 

387 "critical": "error", 

388 } 

389 

390 data = { 

391 "alert_title": alert_title, 

392 "alert_message": alert_message, 

393 "severity": severity.upper(), 

394 "severity_class": severity_map.get(severity.lower(), "warning"), 

395 "component": component, 

396 "occurred_at": datetime.now(UTC).isoformat(), 

397 } 

398 

399 subject, body, html_body = self.template_renderer.render_template( 

400 NotificationType.SYSTEM_ALERT, 

401 data, 

402 ) 

403 

404 notification = Notification( 

405 type=NotificationType.SYSTEM_ALERT, 

406 subject=subject, 

407 body=body, 

408 html_body=html_body, 

409 recipients=recipients, 

410 data=data, 

411 priority="high" if severity.lower() in ("error", "critical") else "normal", 

412 ) 

413 

414 return await self.send(notification) 

415 

416 

417__all__ = ["AdminNotificationService"]