send push notification (#26)

* send push notification

* wip

* fixes

* customize notification body

* add user service

* get device token

* fix uwu

---------

Co-authored-by: violeta <viperez@fi.uba.ar>

Author: feijooso

.env.dist

@@ -13,6 +13,7 @@ DATABASE_URL=
 
 # External services
 PLANT_SERVICE_URL=
+USERS_SERVICE_URL=
 
 # RabbitMQ
 CLOUDAMQP_URL=

.gitignore

@@ -4,4 +4,4 @@ __pycache__/
 /venv/*
 *.lock
 .DS_Store
-vscode/
+vscode/

app/exceptions/deviating_parameters.py

@@ -1,4 +1,5 @@
 class DeviatedParametersError(Exception):
     def __init__(self, parameters):
         self.parameters = parameters
-        super().__init__(f"Out of range parameters: {self.parameters}")
+        super().__init__(f"Out of range parameters: {self.parameters} "
+                         f"type: {type(self.parameters)}")

app/main.py

@@ -15,6 +15,7 @@
 )
 from query_params.QueryParams import DevicePlantQueryParams
 
+
 app = FastAPI()
 repository = MeasurementsRepository()
 service = MeasurementsService(repository)

app/main_rabbitmq.py

@@ -1,3 +1,4 @@
+import asyncio
 import os
 import logging
 import json
@@ -14,9 +15,14 @@ def main():
     logging_level = os.environ.get("LOGGING_LEVEL")
     queue_name = os.environ.get("QUEUE_NAME")
     initialize_log(logging_level)
+    loop = asyncio.get_event_loop()
     consumer = Consumer(queue_name)
-    logger.info("[RABBITMQ] Starting consumer...")
     consumer.run()
+    try:
+        loop.run_forever()
+    finally:
+        loop.close()
+    logger.info("[RABBITMQ] Starting consumer...")
 
 
 def initialize_log(logging_level):

app/schemas/measurement.py

@@ -36,4 +36,3 @@ class MeasurementReadingSchema(Measurement):
     """
     id_device: str
     time_stamp: str
-    device_token: Optional[str]

app/schemas/user.py

@@ -0,0 +1,7 @@
+from pydantic import BaseModel, Field
+
+
+class UserSchema(BaseModel):
+    id: int = Field(...)
+    name: str = Field(...)
+    device_token: str = Field(...)

app/service/rabbitmq/consumer.py

@@ -1,4 +1,6 @@
+import asyncio
 import json
+from ..users import UsersService
 from exceptions.logger_messages import LoggerMessages
 import pydantic
 import logging
@@ -19,6 +21,7 @@
 from os import environ
 from firebase_admin import messaging
 
+
 Base = declarative_base(
     metadata=MetaData(schema=environ.get("POSTGRES_SCHEMA", "measurements_service"))
 )
@@ -34,6 +37,7 @@ def __init__(self, queue_name):
         self.__queue_name = queue_name
         self.__middleware = Middleware()
         self.__sqlAlchemyClient = SQLAlchemyClient()
+        self.__users = UsersService()
 
     def run(self):
         self.__middleware.create_queue(self.__queue_name)
@@ -50,6 +54,9 @@ def obtain_device_plant(self, measurement_from_rabbit):
             logger.error(f"{err} - {type(err)}")
             raise RowNotFoundError(measurement_from_rabbit.id_device, "DEVICE_PLANT")
 
+    async def obtain_user(self, user_id):
+        return await self.__users.get_user(user_id)
+
     def check_package(self, measurement):
         empty_values = []
         if measurement.temperature is None:
@@ -64,22 +71,62 @@ def check_package(self, measurement):
         if measurement.humidity is None:
             empty_values.append("humidity")
 
+        if measurement.id_device is None:
+            empty_values.append("id_device")
+
         if len(empty_values) > 0:
             raise EmptyPackageError(empty_values)
 
-    def send_notification(self, id_user, measurement, error, details):
-        logger.info(LoggerMessages.USER_NOTIFIED.format(id_user))
+    def generate_notification_body(self, error):
+        parameters = {
+            'temperature': 'temperatura',
+            'humidity': 'humedad',
+            'light': 'luz',
+            'watering': 'riego'
+        }
+
+        low_parameters = []
+        high_parameters = []
+        error_dict = error.parameters.dict()
+
+        for param, value in error_dict.items():
+            if value == 'lower':
+                low_parameters.append(parameters.get(param, param))
+            elif value == 'higher':
+                high_parameters.append(parameters.get(param, param))
+
+        low_msg = ", ".join(f"{param}" for param in low_parameters)
+        high_msg = ", ".join(f"{param}" for param in high_parameters)
+
+        if low_msg and high_msg:
+            return (
+                f"Los siguientes parámetros están bajos: {low_msg}. "
+                f"Y los siguientes están altos: {high_msg}."
+            )
+        elif low_msg:
+            return f"Los siguientes parámetros están bajos: {low_msg}."
+        elif high_msg:
+            return f"Los siguientes parámetros están altos: {high_msg}."
 
-        print(f"details: {details}")
-        print(f"error: {error}")
+    async def send_notification(self, id_user, measurement, error, details):
+        device_plant = self.obtain_device_plant(measurement)
 
-        if measurement.device_token is not None:
-            message = messaging.Message(
-                notification=messaging.Notification(title="Estado de tu planta",
-                                                    body="details"),
-                token=measurement.device_token,
-                )
-            messaging.send(message)
+        user = await self.obtain_user(device_plant.id_user)
+
+        notification_body = self.generate_notification_body(error)
+
+        try:
+            if user.device_token is not None:
+                message = messaging.Message(
+                    notification=messaging.Notification(title="Estado de tu planta",
+                                                        body=notification_body),
+                    token=user.device_token)
+                messaging.send(message)
+            logger.info(LoggerMessages.USER_NOTIFIED.format(id_user))
+
+        except Exception as e:
+            logger.info(e)
+            pass
 
     def apply_rules(self, measurement,  device_plant):
         deviated_parameters = apply_rules(measurement, device_plant.plant_type)
@@ -136,19 +183,18 @@ def __callback(self, body):
             )
             logger.debug(LoggerMessages.ERROR_DETAILS.format(err, body))
 
-            device_plant = None  # For not saving the measurement.
+            device_plant = None
         except EmptyPackageError as err:
             logger.warn(LoggerMessages.EMPTY_PACKAGE_RECEIVED)
             logger.debug(LoggerMessages.ERROR_DETAILS.format(err, body))
 
-            self.send_notification(device_plant.id_user, measurement, err, body)
-
-            measurement = None  # For not saving the measurement.
+            measurement = None
         except DeviatedParametersError as err:
             logger.warn(LoggerMessages.DEVIATING_PARAMETERS)
             logger.debug(LoggerMessages.ERROR_DETAILS.format(err, body))
 
-            self.send_notification(device_plant.id_user, measurement, err, body)
+            asyncio.run(self.send_notification(device_plant.id_user, measurement,
+                                               err, body))
 
         if device_plant is not None and measurement is not None:
             try:

app/service/users.py

@@ -0,0 +1,42 @@
+import logging
+from httpx import (
+    AsyncClient,
+    codes,
+    HTTPStatusError
+)
+from os import environ
+from fastapi import HTTPException
+from typing import Optional
+from schemas.user import UserSchema
+
+logger = logging.getLogger("app")
+logger.setLevel("DEBUG")
+
+USERS_SERVICE_URL = environ["USERS_SERVICE_URL"]
+
+
+class UsersService():
+    @staticmethod
+    async def get_user(user_id: int) -> Optional[UserSchema]:
+        try:
+            async with AsyncClient() as client:
+                response = await client.get(
+                    USERS_SERVICE_URL + f"/users/{user_id}"
+                )
+                if response.status_code == codes.OK:
+                    message = response.json()["message"]
+                    return UserSchema(id=message["id"], name=message["name"],
+                                      device_token=message["device_token"])
+                elif response.status_code == codes.NOT_FOUND:
+                    return None
+                else:
+                    return response.raise_for_status().json()
+
+        except HTTPStatusError as e:
+            logger.error(
+                "Users service cannot be accessed because: " + str(e)
+                )
+            raise HTTPException(
+                status_code=e.response.status_code,
+                detail=e.response.content.decode(),
+            )