Coverage for services\push\push_service.py: 79.5%

1import logging 

2from typing import Optional 

3 

4import firebase_admin 

5from decouple import config 

6from django.utils import timezone 

7from firebase_admin import credentials, messaging 

8 

9logger = logging.getLogger(__name__) 

10 

11 

12# Inicializa la app de Firebase la primera vez que se necesita 

13def _ensure_initialized(): 

14 if not firebase_admin._apps: # Solo se inicializa si no hay apps ya inicializadas 

15 cred = credentials.Certificate(config('FIREBASE_CREDENTIALS_PATH')) 

16 firebase_admin.initialize_app(cred) 

17 

18# Clase para enviar notificaciones push a los usuarios 

19class PushNotification: 

20 

21 #Funcion para enviar una notificacion push a todos los dispositivos de un usuario 

22 @staticmethod 

23 def send_to_user(user, title: str, body: str, notification_type: str = "notification", data: Optional[dict] = None): 

24 

25 devices = user.devices.filter(is_deleted=False) 

26 if not devices.exists():# Si no hay dispositivos, no se hace nada 

27 return 

28 try: 

29 _ensure_initialized() # Inicializa Firebase Admin si no estaba inicializado 

30 except Exception as e: 

31 logger.error(f"Could not initialize Firebase Admin: {str(e)}") 

32 return 

33 

34 for device in devices: # Por cada uno de los dispositivos, se envia la notificacion push 

35 message = messaging.Message( 

36 notification=messaging.Notification(title=title, body=body), 

37 data={"type": notification_type, **{k: str(v) for k, v in (data or {}).items()}}, 

38 token=device.fcm_token, 

39 ) 

40 

41 try: 

42 messaging.send(message) 

43 except firebase_admin.exceptions.NotFoundError: 

44 # Si el token de FCM ya no es valido, se marca el dispositivo como borrado 

45 logger.info(f"FCM token invalid, marking device as deleted {device.id}") 

46 device.is_deleted = True 

47 device.deleted_at = timezone.now() 

48 device.save() 

49 except Exception as e: 

50 logger.error(f"Error sending push notification to device {device.id}: {str(e)}")