add user service

Author: feijooso

.env.dist

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

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,5 @@
 import json
+from ..users import UsersService
 from exceptions.logger_messages import LoggerMessages
 import pydantic
 import logging
@@ -34,6 +35,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 +52,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:
@@ -102,11 +107,14 @@ def generate_notification_body(self, error):
             return f"Los siguientes parámetros están altos: {high_msg}."
 
     def send_notification(self, id_user, measurement, error, details):
+        device_plant = self.obtain_device_plant(measurement)
+
+        user = self.obtain_user(device_plant.id_user)
 
         notification_body = self.generate_notification_body(error)
 
         try:
-            if measurement.device_token is not None:
+            if user.device_token is not None:
                 message = messaging.Message(
                     notification=messaging.Notification(title="Estado de tu planta",
                                                         body=notification_body),

app/service/users.py

@@ -0,0 +1,40 @@
+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:
+                    return UserSchema(**response.json())
+                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(),
+            )