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

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

2 

3from __future__ import annotations 

4 

5from datetime import UTC, datetime 

6from typing import 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.contracts.mailer.protocols import MailerProtocol 

19from lexigram.di.decorators import inject 

20from lexigram.result import Err, Ok, Result 

21 

22 

23@inject 

24class AdminNotificationService: 

25 """Service for sending admin notifications. 

26 

27 Integrates with lexigram.messaging for email delivery 

28 and provides admin-specific templates and functionality. 

29 

30 Example: 

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

32 >>> 

33 >>> # Send user created notification 

34 >>> await service.notify_user_created( 

35 ... user=user, 

36 ... created_by=admin, 

37 ... recipients=admin_list, 

38 ... ) 

39 >>> 

40 >>> # Send bulk completion notification 

41 >>> await service.notify_bulk_completed( 

42 ... operation="delete", 

43 ... resource="users", 

44 ... total=100, 

45 ... successful=98, 

46 ... failed=2, 

47 ... recipients=[admin], 

48 ... ) 

49 """ 

50 

51 def __init__( 

52 self, 

53 mailer: MailerProtocol | None = None, 

54 config: AdminNotificationConfig | None = None, 

55 ): 

56 self.config = config or AdminNotificationConfig() 

57 

58 # Initialize components 

59 app_name: str = ( 

60 getattr(self.config, "app_name", None) 

61 or getattr(self.config, "email_from_name", "Admin") 

62 or "Admin" 

63 ) 

64 self.template_renderer = TemplateRenderer(app_name) 

65 self.email_sender = EmailSender( 

66 mailer=mailer, 

67 from_email=self.config.email_from, 

68 from_name=self.config.email_from_name, 

69 ) 

70 

71 self._sent_count = 0 

72 

73 async def send( 

74 self, 

75 notification: Notification, 

76 ) -> Result[NotificationResult, NotificationError]: 

77 """Send a notification. 

78 

79 Args: 

80 notification: Notification to send 

81 

82 Returns: 

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

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

85 """ 

86 enabled_types = getattr(self.config, "enabled_types", None) 

87 if enabled_types and notification.type not in enabled_types: 

88 result = NotificationResult( 

89 notification_id=notification.id, 

90 recipients_sent=0, 

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

92 ) 

93 return Ok(result) 

94 

95 sent = 0 

96 failed = 0 

97 errors: list[str] = [] 

98 

99 for recipient in notification.recipients: 

100 # Check recipient preferences 

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

102 continue 

103 

104 # Send via channels 

105 for channel in notification.channels: 

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

107 try: 

108 await self.email_sender.send_email( 

109 recipient=recipient, 

110 subject=notification.subject, 

111 body=notification.body, 

112 html_body=notification.html_body, 

113 ) 

114 sent += 1 

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

116 failed += 1 

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

118 

119 result = NotificationResult( 

120 notification_id=notification.id, 

121 recipients_sent=sent, 

122 recipients_failed=failed, 

123 errors=errors, 

124 ) 

125 

126 if sent == 0 and failed > 0: 

127 return Err( 

128 NotificationError( 

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

130 ) 

131 ) 

132 

133 return Ok(result) 

134 

135 # ======================================================================== 

136 # Convenience Methods 

137 # ======================================================================== 

138 

139 async def notify_user_created( 

140 self, 

141 user: Any, 

142 created_by: Any, 

143 recipients: list[NotificationRecipient], 

144 ) -> Result[NotificationResult, NotificationError]: 

145 """Send user created notification.""" 

146 data = { 

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

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

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

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

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

152 "user_url": f"{getattr(self.config, 'base_url', '')}/users/{getattr(user, 'id', '')}", 

153 } 

154 

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

156 NotificationType.USER_CREATED, 

157 data, 

158 ) 

159 

160 notification = Notification( 

161 type=NotificationType.USER_CREATED, 

162 subject=subject, 

163 body=body, 

164 html_body=html_body, 

165 recipients=recipients, 

166 data=data, 

167 ) 

168 

169 return await self.send(notification) 

170 

171 async def notify_user_invited( 

172 self, 

173 user_email: str, 

174 user_name: str, 

175 invite_url: str, 

176 expires_in: str = "7 days", 

177 ) -> Result[NotificationResult, NotificationError]: 

178 """Send user invitation notification.""" 

179 data = { 

180 "user_name": user_name, 

181 "user_email": user_email, 

182 "invite_url": invite_url, 

183 "expires_in": expires_in, 

184 } 

185 

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

187 NotificationType.USER_INVITED, 

188 data, 

189 ) 

190 

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

192 

193 notification = Notification( 

194 type=NotificationType.USER_INVITED, 

195 subject=subject, 

196 body=body, 

197 html_body=html_body, 

198 recipients=[recipient], 

199 data=data, 

200 ) 

201 

202 return await self.send(notification) 

203 

204 async def notify_password_reset( 

205 self, 

206 user_email: str, 

207 user_name: str, 

208 reset_url: str, 

209 expires_in: str = "1 hour", 

210 ) -> Result[NotificationResult, NotificationError]: 

211 """Send password reset notification.""" 

212 data = { 

213 "user_name": user_name, 

214 "reset_url": reset_url, 

215 "expires_in": expires_in, 

216 } 

217 

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

219 NotificationType.PASSWORD_RESET, 

220 data, 

221 ) 

222 

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

224 

225 notification = Notification( 

226 type=NotificationType.PASSWORD_RESET, 

227 subject=subject, 

228 body=body, 

229 html_body=html_body, 

230 recipients=[recipient], 

231 data=data, 

232 ) 

233 

234 return await self.send(notification) 

235 

236 async def notify_email_verification( 

237 self, 

238 user_email: str, 

239 user_name: str, 

240 verify_url: str, 

241 expires_in: str = "24 hours", 

242 ) -> Result[NotificationResult, NotificationError]: 

243 """Send email verification notification.""" 

244 data = { 

245 "user_name": user_name, 

246 "verify_url": verify_url, 

247 "expires_in": expires_in, 

248 } 

249 

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

251 NotificationType.EMAIL_VERIFICATION, 

252 data, 

253 ) 

254 

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

256 

257 notification = Notification( 

258 type=NotificationType.EMAIL_VERIFICATION, 

259 subject=subject, 

260 body=body, 

261 html_body=html_body, 

262 recipients=[recipient], 

263 data=data, 

264 ) 

265 

266 return await self.send(notification) 

267 

268 async def notify_email_otp( 

269 self, 

270 user_email: str, 

271 user_name: str, 

272 code: str, 

273 expires_in: str = "10 minutes", 

274 ) -> Result[NotificationResult, NotificationError]: 

275 """Send email one-time-password notification.""" 

276 data = { 

277 "user_name": user_name, 

278 "code": code, 

279 "expires_in": expires_in, 

280 } 

281 

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

283 NotificationType.EMAIL_OTP, 

284 data, 

285 ) 

286 

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

288 

289 notification = Notification( 

290 type=NotificationType.EMAIL_OTP, 

291 subject=subject, 

292 body=body, 

293 html_body=html_body, 

294 recipients=[recipient], 

295 data=data, 

296 ) 

297 

298 return await self.send(notification) 

299 

300 async def notify_bulk_started( 

301 self, 

302 operation_name: str, 

303 resource: str, 

304 total_items: int, 

305 started_by: Any, 

306 recipients: list[NotificationRecipient], 

307 ) -> Result[NotificationResult, NotificationError]: 

308 """Send bulk operation started notification.""" 

309 data = { 

310 "operation_name": operation_name, 

311 "resource": resource, 

312 "total_items": total_items, 

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

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

315 } 

316 

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

318 NotificationType.BULK_STARTED, 

319 data, 

320 ) 

321 

322 notification = Notification( 

323 type=NotificationType.BULK_STARTED, 

324 subject=subject, 

325 body=body, 

326 html_body=html_body, 

327 recipients=recipients, 

328 data=data, 

329 ) 

330 

331 return await self.send(notification) 

332 

333 async def notify_bulk_completed( 

334 self, 

335 operation_name: str, 

336 resource: str, 

337 total_items: int, 

338 successful: int, 

339 failed: int, 

340 duration: str, 

341 recipients: list[NotificationRecipient], 

342 results_url: str | None = None, 

343 ) -> Result[NotificationResult, NotificationError]: 

344 """Send bulk operation completed notification.""" 

345 data = { 

346 "operation_name": operation_name, 

347 "resource": resource, 

348 "total_items": total_items, 

349 "successful": successful, 

350 "failed": failed, 

351 "duration": duration, 

352 "results_url": results_url 

353 or f"{getattr(self.config, 'base_url', '')}/{resource}", 

354 } 

355 

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

357 NotificationType.BULK_COMPLETED, 

358 data, 

359 ) 

360 

361 notification = Notification( 

362 type=NotificationType.BULK_COMPLETED, 

363 subject=subject, 

364 body=body, 

365 html_body=html_body, 

366 recipients=recipients, 

367 data=data, 

368 ) 

369 

370 return await self.send(notification) 

371 

372 async def notify_bulk_failed( 

373 self, 

374 operation_name: str, 

375 resource: str, 

376 error_message: str, 

377 processed: int, 

378 recipients: list[NotificationRecipient], 

379 ) -> Result[NotificationResult, NotificationError]: 

380 """Send bulk operation failed notification.""" 

381 data = { 

382 "operation_name": operation_name, 

383 "resource": resource, 

384 "error_message": error_message, 

385 "processed": processed, 

386 } 

387 

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

389 NotificationType.BULK_FAILED, 

390 data, 

391 ) 

392 

393 notification = Notification( 

394 type=NotificationType.BULK_FAILED, 

395 subject=subject, 

396 body=body, 

397 html_body=html_body, 

398 recipients=recipients, 

399 data=data, 

400 ) 

401 

402 return await self.send(notification) 

403 

404 async def notify_export_ready( 

405 self, 

406 export_name: str, 

407 file_format: str, 

408 record_count: int, 

409 file_size: str, 

410 download_url: str, 

411 recipient: NotificationRecipient, 

412 expires_in: str = "24 hours", 

413 ) -> Result[NotificationResult, NotificationError]: 

414 """Send export ready notification.""" 

415 data = { 

416 "export_name": export_name, 

417 "format": file_format, 

418 "record_count": record_count, 

419 "file_size": file_size, 

420 "download_url": download_url, 

421 "expires_in": expires_in, 

422 } 

423 

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

425 NotificationType.EXPORT_READY, 

426 data, 

427 ) 

428 

429 notification = Notification( 

430 type=NotificationType.EXPORT_READY, 

431 subject=subject, 

432 body=body, 

433 html_body=html_body, 

434 recipients=[recipient], 

435 data=data, 

436 ) 

437 

438 return await self.send(notification) 

439 

440 async def notify_system_alert( 

441 self, 

442 alert_title: str, 

443 alert_message: str, 

444 severity: str, 

445 component: str, 

446 recipients: list[NotificationRecipient], 

447 ) -> Result[NotificationResult, NotificationError]: 

448 """Send system alert notification.""" 

449 severity_map = { 

450 "info": "info", 

451 "warning": "warning", 

452 "error": "error", 

453 "critical": "error", 

454 } 

455 

456 data = { 

457 "alert_title": alert_title, 

458 "alert_message": alert_message, 

459 "severity": severity.upper(), 

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

461 "component": component, 

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

463 } 

464 

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

466 NotificationType.SYSTEM_ALERT, 

467 data, 

468 ) 

469 

470 notification = Notification( 

471 type=NotificationType.SYSTEM_ALERT, 

472 subject=subject, 

473 body=body, 

474 html_body=html_body, 

475 recipients=recipients, 

476 data=data, 

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

478 ) 

479 

480 return await self.send(notification) 

481 

482 

483__all__ = ["AdminNotificationService"]